lopi-ui 0.22.0

TUI dashboard and web API for lopi
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321

    #[tokio::test]
    async fn index_returns_html() {
        let app = test_app().await;
        let resp = get_req(app, "/").await;
        assert_eq!(resp.status(), StatusCode::OK);
        let ct = resp
            .headers()
            .get(axum::http::header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .unwrap_or("");
        assert!(ct.contains("text/html"));
    }

    #[tokio::test]
    async fn sse_endpoint_returns_200() {
        let app = test_app().await;
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/sse")
                    .header("Accept", "text/event-stream")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let ct = resp
            .headers()
            .get(axum::http::header::CONTENT_TYPE)
            .and_then(|v| v.to_str().ok())
            .unwrap_or("");
        assert!(ct.contains("text/event-stream"));
    }

    #[tokio::test]
    async fn stats_has_all_fields() {
        let app = test_app().await;
        let resp = get_req(app, "/api/stats").await;
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert!(json.get("running").is_some());
        assert!(json.get("queued").is_some());
        assert!(json.get("succeeded").is_some());
        assert!(json.get("failed").is_some());
        assert!(json.get("uptime_secs").is_some());
    }

    #[tokio::test]
    async fn create_task_accepts_low_priority() {
        let app = test_app().await;
        let body = serde_json::to_string(&serde_json::json!({
            "goal": "low priority maintenance",
            "priority": "low",
        }))
        .unwrap();
        let resp = send_req(app, "POST", "/api/tasks", Some(body)).await;
        assert_eq!(resp.status(), StatusCode::CREATED);
    }

    #[tokio::test]
    async fn create_task_accepts_normal_priority() {
        let app = test_app().await;
        let body = serde_json::to_string(&serde_json::json!({
            "goal": "normal priority work",
            "priority": "normal",
        }))
        .unwrap();
        let resp = send_req(app, "POST", "/api/tasks", Some(body)).await;
        assert_eq!(resp.status(), StatusCode::CREATED);
    }

    #[tokio::test]
    async fn tasks_list_response_has_tasks_array() {
        let app = test_app().await;
        let resp = get_req(app, "/api/tasks").await;
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert!(json["tasks"].is_array());
    }

    #[tokio::test]
    async fn metrics_has_all_metric_names() {
        let app = test_app().await;
        let resp = get_req(app, "/metrics").await;
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let body = String::from_utf8_lossy(&bytes);
        assert!(body.contains("lopi_agents_running"));
        assert!(body.contains("lopi_agents_queued"));
        assert!(body.contains("lopi_tasks_succeeded_total"));
        assert!(body.contains("lopi_tasks_failed_total"));
        assert!(body.contains("lopi_uptime_seconds"));
        assert!(body.contains("lopi_audit_log_total"));
    }

    #[tokio::test]
    async fn health_response_has_service_field() {
        let app = test_app().await;
        let resp = get_req(app, "/api/health").await;
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(json["status"], "ok");
        assert_eq!(json["service"], "lopi");
    }

    #[tokio::test]
    async fn create_task_response_has_queued_field() {
        let app = test_app().await;
        let body = serde_json::to_string(&serde_json::json!({
            "goal": "check queued field",
        }))
        .unwrap();
        let resp = send_req(app, "POST", "/api/tasks", Some(body)).await;
        assert_eq!(resp.status(), StatusCode::CREATED);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert!(json.get("queued").is_some());
        assert!(json.get("id").is_some());
        assert!(json.get("goal").is_some());
    }

    #[tokio::test]
    async fn create_task_with_repo_returns_201() {
        let app = test_app().await;
        let body = serde_json::to_string(&serde_json::json!({
            "goal": "task with repo override",
            "repo": "/tmp/myrepo",
        }))
        .unwrap();
        let resp = send_req(app, "POST", "/api/tasks", Some(body)).await;
        assert_eq!(resp.status(), StatusCode::CREATED);
    }

    #[tokio::test]
    async fn duplicate_task_returns_201_with_duplicate_of() {
        let app = test_app().await;
        let goal = "deduplicated task goal unique xyz";
        let body = serde_json::to_string(&serde_json::json!({
            "goal": goal,
        }))
        .unwrap();
        // Submit first time
        let resp1 = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/api/tasks")
                    .header("Content-Type", "application/json")
                    .body(Body::from(body.clone()))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp1.status(), StatusCode::CREATED);
        let bytes1 = axum::body::to_bytes(resp1.into_body(), usize::MAX)
            .await
            .unwrap();
        let json1: serde_json::Value = serde_json::from_slice(&bytes1).unwrap();
        assert_eq!(json1["queued"], true);

        // Submit same goal again — should be deduplicated
        let resp2 = send_req(app, "POST", "/api/tasks", Some(body)).await;
        assert_eq!(resp2.status(), StatusCode::CREATED);
        let bytes2 = axum::body::to_bytes(resp2.into_body(), usize::MAX)
            .await
            .unwrap();
        let json2: serde_json::Value = serde_json::from_slice(&bytes2).unwrap();
        // duplicate_of is set when the task already exists
        assert!(json2["duplicate_of"].is_string());
    }

    // ─── F2 — per-task SSE stream + log ring buffer ──────────────────

    /// `GET /api/tasks/:id/logs` on a *known* task with no logs yet returns
    /// 200 with an empty array — "no logs" is a valid state. (An *unknown*
    /// id is 404, gated on task existence — see `f8_id_scoped_reads_status_codes`.)
    #[tokio::test]
    async fn task_logs_known_task_no_logs_returns_empty_array() {
        let (app, store) = test_app_with_store().await;
        let task = Task::new("known task, no logs yet");
        store.save_task(&task, "running").await.unwrap();
        let tid = task.id.0.to_string();

        let resp = get_req(app, &format!("/api/tasks/{tid}/logs")).await;
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(json["task_id"], tid);
        assert_eq!(json["logs"].as_array().unwrap().len(), 0);
    }

    /// Seed a couple of log rows via the store for a saved task, then read
    /// them back via the endpoint to verify the wire shape. The task is saved
    /// first because `get_logs` gates on task existence (Verify-1 F8).
    #[tokio::test]
    async fn task_logs_returns_seeded_rows_oldest_first() {
        let (app, store) = test_app_with_store().await;

        let task = Task::new("task with logs");
        store.save_task(&task, "running").await.unwrap();
        let tid = task.id.0.to_string();
        let now = chrono::Utc::now();
        for (i, level) in ["info", "warn", "info"].iter().enumerate() {
            store
                .record_task_log(&tid, now, level, &format!("line {i}"))
                .await
                .unwrap();
        }

        let resp = get_req(app, &format!("/api/tasks/{tid}/logs?n=10")).await;
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        let logs = json["logs"].as_array().unwrap();
        assert_eq!(logs.len(), 3);
        assert_eq!(logs[0]["line"], "line 0");
        assert_eq!(logs[1]["level"], "warn");
        assert_eq!(logs[2]["line"], "line 2");
    }

    /// `GET /api/logs` on a fresh store returns 200 with an empty array.
    #[tokio::test]
    async fn global_logs_empty_store_returns_empty_array() {
        let app = test_app().await;
        let resp = get_req(app, "/api/logs").await;
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(json["logs"].as_array().unwrap().len(), 0);
    }

    /// Seed rows across two tasks, then read the global tail and verify
    /// it interleaves tasks oldest-first with the wire shape intact.
    #[tokio::test]
    async fn global_logs_returns_rows_across_tasks_oldest_first() {
        let (app, store) = test_app_with_store().await;

        let now = chrono::Utc::now();
        store.record_task_log("t-a", now, "info", "a1").await.unwrap();
        store.record_task_log("t-b", now, "error", "b1").await.unwrap();
        store.record_task_log("t-a", now, "info", "a2").await.unwrap();

        let resp = get_req(app, "/api/logs?n=2").await;
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        let logs = json["logs"].as_array().unwrap();
        // n=2 → newest window (b1, a2) in chronological order.
        assert_eq!(logs.len(), 2);
        assert_eq!(logs[0]["task_id"], "t-b");
        assert_eq!(logs[0]["level"], "error");
        assert_eq!(logs[1]["line"], "a2");
    }

    // ─── F3 — per-agent rate limiting ────────────────────────────────

    /// Posting a rate limit with `max_per_minute: 0` returns 422.
    #[tokio::test]
    async fn agent_rate_limit_zero_per_minute_returns_422() {
        let app = test_app().await;
        let body = serde_json::to_string(&serde_json::json!({
            "max_per_minute": 0,
            "max_concurrent": 4,
        }))
        .unwrap();
        let resp = send_req(app, "POST", "/api/agents/alpha/rate-limit", Some(body)).await;
        assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY);
    }

    /// Register → GET → DELETE → GET round trip for a per-agent limit.
    #[tokio::test]
    async fn agent_rate_limit_round_trip_register_get_delete() {
        let app = test_app().await;
        // 1. Register.
        let body = serde_json::to_string(&serde_json::json!({
            "max_per_minute": 30,
            "max_concurrent": 2,
        }))
        .unwrap();
        let resp = send_req(app.clone(), "POST", "/api/agents/alpha/rate-limit", Some(body)).await;
        assert_eq!(resp.status(), StatusCode::CREATED);
        // 2. GET.
        let resp = get_req(app.clone(), "/api/agents/alpha/rate-limit").await;
        assert_eq!(resp.status(), StatusCode::OK);
        let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        assert_eq!(json["max_per_minute"], 30);
        assert_eq!(json["max_concurrent"], 2);
        assert_eq!(json["in_flight"], 0);
        // 3. DELETE.
        let resp = send_req(app.clone(), "DELETE", "/api/agents/alpha/rate-limit", None).await;
        assert_eq!(resp.status(), StatusCode::OK);
        // 4. GET after delete → 404.
        let resp = get_req(app, "/api/agents/alpha/rate-limit").await;
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }