[WIP] Heartbeat checkpoint - 2026-01-31 22:48:28 UTC

2 files changed, 333 insertions, 0 deletions

diff --git a/makima/src/server/handlers/mesh_red_team.rs b/makima/src/server/handlers/mesh_red_team.rs
index 6031c2c..1d8e0b0 100644
--- a/makima/src/server/handlers/mesh_red_team.rs
+++ b/makima/src/server/handlers/mesh_red_team.rs
@@ -501,3 +501,334 @@ pub async fn get_status(
     )
         .into_response()
 }
+
+// =============================================================================
+// Phase 2: Task Output Subscription & Diff Access
+// =============================================================================
+
+/// Response for task output subscription.
+#[derive(Debug, Serialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct TaskOutputSubscriptionResponse {
+    /// Whether subscription was successful
+    pub success: bool,
+    /// Contract ID being monitored
+    pub contract_id: Uuid,
+    /// Message about the subscription
+    pub message: String,
+    /// List of work tasks currently active in the contract
+    pub active_tasks: Vec<MonitoredTaskInfo>,
+}
+
+/// Information about a task being monitored.
+#[derive(Debug, Serialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct MonitoredTaskInfo {
+    pub task_id: Uuid,
+    pub name: String,
+    pub status: String,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+}
+
+/// Response for red team diff request.
+#[derive(Debug, Serialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct RedTeamDiffResponse {
+    /// Task ID that was queried
+    pub task_id: Uuid,
+    /// Whether the request was successful
+    pub success: bool,
+    /// The diff content (if available)
+    pub diff: Option<String>,
+    /// Error message if any
+    pub error: Option<String>,
+}
+
+/// Get list of active work tasks that the red team can monitor.
+///
+/// GET /api/v1/mesh/red-team/monitored-tasks
+///
+/// Returns information about all active work tasks in the contract that
+/// the red team can monitor. This helps the red team understand what
+/// tasks are running and can be analyzed.
+#[utoipa::path(
+    get,
+    path = "/api/v1/mesh/red-team/monitored-tasks",
+    responses(
+        (status = 200, description = "List of monitored tasks", body = TaskOutputSubscriptionResponse),
+        (status = 401, description = "Unauthorized - tool key required"),
+        (status = 403, description = "Forbidden - not a red team task"),
+        (status = 404, description = "Task not found"),
+        (status = 503, description = "Database not available"),
+        (status = 500, description = "Internal server error"),
+    ),
+    security(
+        ("tool_key" = [])
+    ),
+    tag = "Mesh Red Team"
+)]
+pub async fn get_monitored_tasks(
+    State(state): State<SharedState>,
+    headers: HeaderMap,
+) -> impl IntoResponse {
+    let (_red_team_task_id, owner_id, contract_id) =
+        match verify_red_team_auth(&state, &headers).await {
+            Ok(ids) => ids,
+            Err(e) => return e.into_response(),
+        };
+
+    let pool = state.db_pool.as_ref().unwrap();
+
+    // Get all tasks in the contract that are not the red team task and not supervisors
+    let tasks = match sqlx::query_as::<_, (Uuid, String, String, chrono::DateTime<chrono::Utc>)>(
+        r#"
+        SELECT t.id, t.name, t.status, t.created_at
+        FROM tasks t
+        WHERE t.contract_id = $1
+          AND t.owner_id = $2
+          AND t.is_supervisor = FALSE
+          AND (t.name NOT ILIKE '%red-team%' AND t.name NOT ILIKE '%red_team%' AND t.name NOT ILIKE '%redteam%')
+          AND t.status NOT IN ('done', 'cancelled', 'failed')
+        ORDER BY t.created_at DESC
+        "#,
+    )
+    .bind(contract_id)
+    .bind(owner_id)
+    .fetch_all(pool)
+    .await
+    {
+        Ok(tasks) => tasks,
+        Err(e) => {
+            tracing::error!(error = %e, "Failed to get monitored tasks");
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(ApiError::new("DB_ERROR", "Failed to get monitored tasks")),
+            )
+                .into_response();
+        }
+    };
+
+    let active_tasks: Vec<MonitoredTaskInfo> = tasks
+        .into_iter()
+        .map(|(task_id, name, status, created_at)| MonitoredTaskInfo {
+            task_id,
+            name,
+            status,
+            created_at,
+        })
+        .collect();
+
+    let task_count = active_tasks.len();
+
+    (
+        StatusCode::OK,
+        Json(TaskOutputSubscriptionResponse {
+            success: true,
+            contract_id,
+            message: format!(
+                "Found {} active work task(s) to monitor. Task outputs are broadcast in real-time.",
+                task_count
+            ),
+            active_tasks,
+        }),
+    )
+        .into_response()
+}
+
+/// Get the diff for a specific task (read-only access for red team).
+///
+/// GET /api/v1/mesh/red-team/tasks/{task_id}/diff
+///
+/// Red team tasks can use this endpoint to get the diff of changes made by
+/// a work task. This is read-only access - the red team cannot modify anything.
+#[utoipa::path(
+    get,
+    path = "/api/v1/mesh/red-team/tasks/{task_id}/diff",
+    params(
+        ("task_id" = Uuid, Path, description = "Task ID to get diff for")
+    ),
+    responses(
+        (status = 200, description = "Task diff", body = RedTeamDiffResponse),
+        (status = 401, description = "Unauthorized - tool key required"),
+        (status = 403, description = "Forbidden - not a red team task or task not in same contract"),
+        (status = 404, description = "Task not found"),
+        (status = 503, description = "Database not available or daemon unavailable"),
+        (status = 500, description = "Internal server error"),
+    ),
+    security(
+        ("tool_key" = [])
+    ),
+    tag = "Mesh Red Team"
+)]
+pub async fn get_task_diff(
+    State(state): State<SharedState>,
+    Path(task_id): Path<Uuid>,
+    headers: HeaderMap,
+) -> impl IntoResponse {
+    let (_red_team_task_id, owner_id, contract_id) =
+        match verify_red_team_auth(&state, &headers).await {
+            Ok(ids) => ids,
+            Err(e) => return e.into_response(),
+        };
+
+    let pool = state.db_pool.as_ref().unwrap();
+
+    // Get the target task and verify it's in the same contract
+    let task = match repository::get_task_for_owner(pool, task_id, owner_id).await {
+        Ok(Some(t)) => t,
+        Ok(None) => {
+            return (
+                StatusCode::NOT_FOUND,
+                Json(ApiError::new("NOT_FOUND", "Task not found")),
+            )
+                .into_response();
+        }
+        Err(e) => {
+            tracing::error!(error = %e, "Failed to get task");
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(ApiError::new("DB_ERROR", "Failed to get task")),
+            )
+                .into_response();
+        }
+    };
+
+    // Verify the task belongs to the same contract
+    if task.contract_id != Some(contract_id) {
+        return (
+            StatusCode::FORBIDDEN,
+            Json(ApiError::new(
+                "WRONG_CONTRACT",
+                "Red team can only access tasks in its own contract",
+            )),
+        )
+            .into_response();
+    }
+
+    // Red team should not be able to get diffs of other red team tasks or supervisors
+    let is_red_team = task.name.to_lowercase().contains("red-team")
+        || task.name.to_lowercase().contains("red_team")
+        || task.name.to_lowercase().contains("redteam");
+
+    if is_red_team {
+        return (
+            StatusCode::FORBIDDEN,
+            Json(ApiError::new(
+                "CANNOT_MONITOR_RED_TEAM",
+                "Red team cannot monitor other red team tasks",
+            )),
+        )
+            .into_response();
+    }
+
+    if task.is_supervisor {
+        return (
+            StatusCode::FORBIDDEN,
+            Json(ApiError::new(
+                "CANNOT_MONITOR_SUPERVISOR",
+                "Red team cannot get diff from supervisor tasks",
+            )),
+        )
+            .into_response();
+    }
+
+    // Get daemon running the task
+    let Some(daemon_id) = task.daemon_id else {
+        return (
+            StatusCode::OK,
+            Json(RedTeamDiffResponse {
+                task_id,
+                success: false,
+                diff: None,
+                error: Some("Task has no assigned daemon - task may be pending or completed".to_string()),
+            }),
+        )
+            .into_response();
+    };
+
+    // Send GetTaskDiff command to daemon
+    // Note: This is the same command supervisors use, but red team access is read-only
+    let cmd = DaemonCommand::GetTaskDiff { task_id };
+
+    if let Err(e) = state.send_daemon_command(daemon_id, cmd).await {
+        tracing::warn!(
+            error = %e,
+            task_id = %task_id,
+            daemon_id = %daemon_id,
+            "Failed to send GetTaskDiff command for red team"
+        );
+        return (
+            StatusCode::OK,
+            Json(RedTeamDiffResponse {
+                task_id,
+                success: false,
+                diff: None,
+                error: Some(format!("Failed to send command to daemon: {}", e)),
+            }),
+        )
+            .into_response();
+    }
+
+    // Log this access for audit purposes
+    tracing::info!(
+        red_team_task_id = %_red_team_task_id,
+        target_task_id = %task_id,
+        contract_id = %contract_id,
+        "Red team requested diff for task (read-only access)"
+    );
+
+    (
+        StatusCode::OK,
+        Json(RedTeamDiffResponse {
+            task_id,
+            success: true,
+            diff: None, // Diff will be streamed via daemon response
+            error: Some("Diff command sent - response will be streamed".to_string()),
+        }),
+    )
+        .into_response()
+}
+
+/// Filter task output notifications for a specific contract.
+///
+/// This helper function can be used by the red team to filter task output
+/// notifications to only include outputs from work tasks in its contract.
+pub fn filter_task_output_for_contract(
+    notification: &TaskOutputNotification,
+    contract_id: Uuid,
+    owner_id: Uuid,
+    pool: &sqlx::PgPool,
+) -> bool {
+    // Note: This is a placeholder for the actual filtering logic.
+    // In practice, the filtering would be done at the subscription level
+    // by querying the database to verify the task belongs to the contract.
+    // The broadcast channel sends all task outputs, so filtering must be done
+    // by the subscriber.
+
+    // The owner_id check ensures data isolation
+    if notification.owner_id != Some(owner_id) {
+        return false;
+    }
+
+    // Additional filtering would need to be done by looking up the task
+    // to verify it belongs to the contract. This is handled in the
+    // WebSocket subscription handler.
+    true
+}
+
+/// Information about a red team notification for logging/history.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct RedTeamNotificationRecord {
+    pub notification_id: Uuid,
+    pub red_team_task_id: Uuid,
+    pub contract_id: Uuid,
+    pub supervisor_task_id: Option<Uuid>,
+    pub severity: String,
+    pub message: String,
+    pub related_task_id: Option<Uuid>,
+    pub file_path: Option<String>,
+    pub context: Option<String>,
+    pub delivered: bool,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+}
diff --git a/makima/src/server/mod.rs b/makima/src/server/mod.rs
index 8456006..34f8bb9 100644
--- a/makima/src/server/mod.rs
+++ b/makima/src/server/mod.rs
@@ -135,6 +135,8 @@ pub fn make_router(state: SharedState) -> Router {
         // Red team endpoints (for red team tasks to notify supervisors)
         .route("/mesh/red-team/notify", post(mesh_red_team::notify_supervisor))
         .route("/mesh/red-team/status", get(mesh_red_team::get_status))
+        .route("/mesh/red-team/monitored-tasks", get(mesh_red_team::get_monitored_tasks))
+        .route("/mesh/red-team/tasks/{task_id}/diff", get(mesh_red_team::get_task_diff))
         // Mesh WebSocket endpoints
         .route("/mesh/tasks/subscribe", get(mesh_ws::task_subscription_handler))
         .route("/mesh/daemons/connect", get(mesh_daemon::daemon_handler))