codetether_agent/server/
mod.rs

1//! HTTP Server
2//!
3//! Main API server for the CodeTether Agent
4
5pub mod auth;
6pub mod policy;
7
8use crate::a2a;
9use crate::audit::{self, AuditCategory, AuditLog, AuditOutcome};
10use crate::bus::AgentBus;
11use crate::cli::ServeArgs;
12use crate::cognition::{
13    AttentionItem, CognitionRuntime, CognitionStatus, CreatePersonaRequest, GlobalWorkspace,
14    LineageGraph, MemorySnapshot, Proposal, ReapPersonaRequest, ReapPersonaResponse,
15    SpawnPersonaRequest, StartCognitionRequest, StopCognitionRequest, beliefs::Belief,
16    executor::DecisionReceipt,
17};
18use crate::config::Config;
19use crate::k8s::K8sManager;
20use crate::tool::{PluginManifest, SigningKey, hash_bytes, hash_file};
21use anyhow::Result;
22use auth::AuthState;
23use axum::{
24    Router,
25    body::Body,
26    extract::Path,
27    extract::{Query, State},
28    http::{Request, StatusCode},
29    middleware::{self, Next},
30    response::sse::{Event, KeepAlive, Sse},
31    response::{Json, Response},
32    routing::{get, post},
33};
34use futures::stream;
35use serde::{Deserialize, Serialize};
36use std::convert::Infallible;
37use std::sync::Arc;
38use tower_http::cors::{AllowHeaders, AllowMethods, AllowOrigin, CorsLayer};
39use tower_http::trace::TraceLayer;
40
41/// Server state shared across handlers
42#[derive(Clone)]
43pub struct AppState {
44    pub config: Arc<Config>,
45    pub cognition: Arc<CognitionRuntime>,
46    pub audit_log: AuditLog,
47    pub k8s: Arc<K8sManager>,
48    pub auth: AuthState,
49    pub bus: Arc<AgentBus>,
50}
51
52/// Audit middleware — logs every request/response to the audit trail.
53async fn audit_middleware(
54    State(state): State<AppState>,
55    request: Request<Body>,
56    next: Next,
57) -> Response {
58    let method = request.method().clone();
59    let path = request.uri().path().to_string();
60    let started = std::time::Instant::now();
61
62    let response = next.run(request).await;
63
64    let duration_ms = started.elapsed().as_millis() as u64;
65    let status = response.status().as_u16();
66    let outcome = if status < 400 {
67        AuditOutcome::Success
68    } else if status == 401 || status == 403 {
69        AuditOutcome::Denied
70    } else {
71        AuditOutcome::Failure
72    };
73
74    state
75        .audit_log
76        .record(audit::AuditEntry {
77            id: uuid::Uuid::new_v4().to_string(),
78            timestamp: chrono::Utc::now(),
79            category: AuditCategory::Api,
80            action: format!("{} {}", method, path),
81            principal: None,
82            outcome,
83            detail: Some(serde_json::json!({ "status": status })),
84            duration_ms: Some(duration_ms),
85        })
86        .await;
87
88    response
89}
90
91/// Mapping from (path pattern, HTTP method) → OPA permission action.
92/// The first matching rule wins.
93struct PolicyRule {
94    pattern: &'static str,
95    methods: Option<&'static [&'static str]>,
96    permission: &'static str,
97}
98
99const POLICY_RULES: &[PolicyRule] = &[
100    // Public / exempt
101    PolicyRule {
102        pattern: "/health",
103        methods: None,
104        permission: "",
105    },
106    PolicyRule {
107        pattern: "/a2a/",
108        methods: None,
109        permission: "",
110    },
111    // K8s management — admin only
112    PolicyRule {
113        pattern: "/v1/k8s/scale",
114        methods: Some(&["POST"]),
115        permission: "admin:access",
116    },
117    PolicyRule {
118        pattern: "/v1/k8s/restart",
119        methods: Some(&["POST"]),
120        permission: "admin:access",
121    },
122    PolicyRule {
123        pattern: "/v1/k8s/",
124        methods: Some(&["GET"]),
125        permission: "admin:access",
126    },
127    // K8s sub-agent lifecycle — admin only
128    PolicyRule {
129        pattern: "/v1/k8s/subagent",
130        methods: Some(&["POST", "DELETE"]),
131        permission: "admin:access",
132    },
133    // Plugin registry — read
134    PolicyRule {
135        pattern: "/v1/plugins",
136        methods: Some(&["GET"]),
137        permission: "agent:read",
138    },
139    // Audit — admin
140    PolicyRule {
141        pattern: "/v1/audit",
142        methods: None,
143        permission: "admin:access",
144    },
145    // Cognition — write operations
146    PolicyRule {
147        pattern: "/v1/cognition/start",
148        methods: Some(&["POST"]),
149        permission: "agent:execute",
150    },
151    PolicyRule {
152        pattern: "/v1/cognition/stop",
153        methods: Some(&["POST"]),
154        permission: "agent:execute",
155    },
156    PolicyRule {
157        pattern: "/v1/cognition/",
158        methods: Some(&["GET"]),
159        permission: "agent:read",
160    },
161    // Swarm persona lifecycle
162    PolicyRule {
163        pattern: "/v1/swarm/personas",
164        methods: Some(&["POST"]),
165        permission: "agent:execute",
166    },
167    PolicyRule {
168        pattern: "/v1/swarm/",
169        methods: Some(&["POST"]),
170        permission: "agent:execute",
171    },
172    PolicyRule {
173        pattern: "/v1/swarm/",
174        methods: Some(&["GET"]),
175        permission: "agent:read",
176    },
177    // Session management
178    PolicyRule {
179        pattern: "/api/session",
180        methods: Some(&["POST"]),
181        permission: "sessions:write",
182    },
183    PolicyRule {
184        pattern: "/api/session/",
185        methods: Some(&["POST"]),
186        permission: "sessions:write",
187    },
188    PolicyRule {
189        pattern: "/api/session",
190        methods: Some(&["GET"]),
191        permission: "sessions:read",
192    },
193    // Config, version, providers, agents — read
194    PolicyRule {
195        pattern: "/api/version",
196        methods: None,
197        permission: "agent:read",
198    },
199    PolicyRule {
200        pattern: "/api/config",
201        methods: None,
202        permission: "agent:read",
203    },
204    PolicyRule {
205        pattern: "/api/provider",
206        methods: None,
207        permission: "agent:read",
208    },
209    PolicyRule {
210        pattern: "/api/agent",
211        methods: None,
212        permission: "agent:read",
213    },
214];
215
216/// Find the required permission for a given path + method.
217/// Returns `Some("")` for exempt, `Some(perm)` for required, `None` for unmatched (pass-through).
218fn match_policy_rule(path: &str, method: &str) -> Option<&'static str> {
219    for rule in POLICY_RULES {
220        let matches = if rule.pattern.ends_with('/') {
221            path.starts_with(rule.pattern) || path == &rule.pattern[..rule.pattern.len() - 1]
222        } else {
223            path == rule.pattern || path.starts_with(&format!("{}/", rule.pattern))
224        };
225        if matches {
226            if let Some(allowed_methods) = rule.methods {
227                if !allowed_methods.contains(&method) {
228                    continue;
229                }
230            }
231            return Some(rule.permission);
232        }
233    }
234    None
235}
236
237/// Policy authorization middleware for Axum.
238///
239/// Maps request paths to OPA permission strings and enforces authorization.
240/// Runs after `require_auth` so the bearer token is already validated.
241/// Currently maps the static bearer token to an admin role since
242/// codetether-agent uses a single shared token model.
243async fn policy_middleware(request: Request<Body>, next: Next) -> Result<Response, StatusCode> {
244    let path = request.uri().path().to_string();
245    let method = request.method().as_str().to_string();
246
247    let permission = match match_policy_rule(&path, &method) {
248        None | Some("") => return Ok(next.run(request).await),
249        Some(perm) => perm,
250    };
251
252    // The current auth model uses a single static token for all access.
253    // When this is the case, the authenticated user effectively has admin role.
254    // Future: extract user claims from JWT and build a proper PolicyUser.
255    let user = policy::PolicyUser {
256        user_id: "bearer-token-user".to_string(),
257        roles: vec!["admin".to_string()],
258        tenant_id: None,
259        scopes: vec![],
260        auth_source: "static_token".to_string(),
261    };
262
263    if let Err(status) = policy::enforce_policy(&user, permission, None).await {
264        tracing::warn!(
265            path = %path,
266            method = %method,
267            permission = %permission,
268            "Policy middleware denied request"
269        );
270        return Err(status);
271    }
272
273    Ok(next.run(request).await)
274}
275
276/// Start the HTTP server
277pub async fn serve(args: ServeArgs) -> Result<()> {
278    let t0 = std::time::Instant::now();
279    tracing::info!("[startup] begin");
280    let config = Config::load().await?;
281    tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] config loaded");
282    let mut cognition = CognitionRuntime::new_from_env();
283    tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] cognition runtime created");
284
285    // Set up tool registry for cognition execution engine.
286    cognition.set_tools(Arc::new(crate::tool::ToolRegistry::with_defaults()));
287    tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] tools registered");
288    let cognition = Arc::new(cognition);
289
290    // Initialize audit log.
291    let audit_log = AuditLog::from_env();
292    let _ = audit::init_audit_log(audit_log.clone());
293
294    // Initialize K8s manager.
295    tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] pre-k8s");
296    let k8s = Arc::new(K8sManager::new().await);
297    tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] k8s done");
298    if k8s.is_available() {
299        tracing::info!("K8s self-deployment enabled");
300    }
301
302    // Initialize mandatory auth.
303    let auth_state = AuthState::from_env();
304    tracing::info!(
305        token_len = auth_state.token().len(),
306        "Auth is mandatory. Token required for all API endpoints."
307    );
308    tracing::info!(
309        audit_entries = audit_log.count().await,
310        "Audit log initialized"
311    );
312
313    // Create agent bus for in-process communication
314    let bus = AgentBus::new().into_arc();
315    tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] bus created");
316
317    if cognition.is_enabled() && env_bool("CODETETHER_COGNITION_AUTO_START", true) {
318        tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] auto-starting cognition");
319        if let Err(error) = cognition.start(None).await {
320            tracing::warn!(%error, "Failed to auto-start cognition loop");
321        } else {
322            tracing::info!("Perpetual cognition auto-started");
323        }
324    }
325
326    tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] building routes");
327    let addr = format!("{}:{}", args.hostname, args.port);
328
329    // Build the agent card
330    let agent_card = a2a::server::A2AServer::default_card(&format!("http://{}", addr));
331    let a2a_server = a2a::server::A2AServer::new(agent_card.clone());
332
333    // Build A2A router separately
334    let a2a_router = a2a_server.router();
335
336    // Start gRPC transport on a separate port
337    let grpc_port = std::env::var("CODETETHER_GRPC_PORT")
338        .ok()
339        .and_then(|p| p.parse::<u16>().ok())
340        .unwrap_or(50051);
341    let grpc_addr: std::net::SocketAddr = format!("{}:{}", args.hostname, grpc_port).parse()?;
342    let grpc_store = crate::a2a::grpc::GrpcTaskStore::with_bus(agent_card, bus.clone());
343    let grpc_service = grpc_store.into_service();
344    tokio::spawn(async move {
345        tracing::info!("gRPC A2A server listening on {}", grpc_addr);
346        if let Err(e) = tonic::transport::Server::builder()
347            .add_service(grpc_service)
348            .serve(grpc_addr)
349            .await
350        {
351            tracing::error!("gRPC server error: {}", e);
352        }
353    });
354
355    let state = AppState {
356        config: Arc::new(config),
357        cognition,
358        audit_log,
359        k8s,
360        auth: auth_state.clone(),
361        bus,
362    };
363
364    let app = Router::new()
365        // Health check (public — auth exempt)
366        .route("/health", get(health))
367        // API routes
368        .route("/api/version", get(get_version))
369        .route("/api/session", get(list_sessions).post(create_session))
370        .route("/api/session/{id}", get(get_session))
371        .route("/api/session/{id}/prompt", post(prompt_session))
372        .route("/api/config", get(get_config))
373        .route("/api/provider", get(list_providers))
374        .route("/api/agent", get(list_agents))
375        // Perpetual cognition APIs
376        .route("/v1/cognition/start", post(start_cognition))
377        .route("/v1/cognition/stop", post(stop_cognition))
378        .route("/v1/cognition/status", get(get_cognition_status))
379        .route("/v1/cognition/stream", get(stream_cognition))
380        .route("/v1/cognition/snapshots/latest", get(get_latest_snapshot))
381        // Swarm persona lifecycle APIs
382        .route("/v1/swarm/personas", post(create_persona))
383        .route("/v1/swarm/personas/{id}/spawn", post(spawn_persona))
384        .route("/v1/swarm/personas/{id}/reap", post(reap_persona))
385        .route("/v1/swarm/lineage", get(get_swarm_lineage))
386        // Belief, attention, governance, workspace APIs
387        .route("/v1/cognition/beliefs", get(list_beliefs))
388        .route("/v1/cognition/beliefs/{id}", get(get_belief))
389        .route("/v1/cognition/attention", get(list_attention))
390        .route("/v1/cognition/proposals", get(list_proposals))
391        .route(
392            "/v1/cognition/proposals/{id}/approve",
393            post(approve_proposal),
394        )
395        .route("/v1/cognition/receipts", get(list_receipts))
396        .route("/v1/cognition/workspace", get(get_workspace))
397        .route("/v1/cognition/governance", get(get_governance))
398        .route("/v1/cognition/personas/{id}", get(get_persona))
399        // Audit trail API
400        .route("/v1/audit", get(list_audit_entries))
401        // K8s self-deployment APIs
402        .route("/v1/k8s/status", get(get_k8s_status))
403        .route("/v1/k8s/scale", post(k8s_scale))
404        .route("/v1/k8s/restart", post(k8s_restart))
405        .route("/v1/k8s/pods", get(k8s_list_pods))
406        .route("/v1/k8s/actions", get(k8s_actions))
407        .route("/v1/k8s/subagent", post(k8s_spawn_subagent))
408        .route(
409            "/v1/k8s/subagent/{id}",
410            axum::routing::delete(k8s_delete_subagent),
411        )
412        // Plugin registry API
413        .route("/v1/plugins", get(list_plugins))
414        .with_state(state.clone())
415        // A2A routes (nested to work with different state type)
416        .nest("/a2a", a2a_router)
417        // Mandatory auth middleware — applies to all routes
418        .layer(middleware::from_fn_with_state(
419            state.clone(),
420            audit_middleware,
421        ))
422        .layer(axum::Extension(state.auth.clone()))
423        .layer(middleware::from_fn(policy_middleware))
424        .layer(middleware::from_fn(auth::require_auth))
425        // CORS + tracing
426        .layer(
427            CorsLayer::new()
428                .allow_origin(AllowOrigin::mirror_request())
429                .allow_credentials(true)
430                .allow_methods(AllowMethods::mirror_request())
431                .allow_headers(AllowHeaders::mirror_request()),
432        )
433        .layer(TraceLayer::new_for_http());
434
435    tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] router built, binding");
436    let listener = tokio::net::TcpListener::bind(&addr).await?;
437    tracing::info!(elapsed_ms = t0.elapsed().as_millis(), "[startup] listening on http://{}", addr);
438
439    axum::serve(listener, app).await?;
440
441    Ok(())
442}
443
444/// Health check response
445async fn health() -> &'static str {
446    "ok"
447}
448
449/// Version info
450#[derive(Serialize)]
451struct VersionInfo {
452    version: &'static str,
453    name: &'static str,
454    binary_hash: Option<String>,
455}
456
457async fn get_version() -> Json<VersionInfo> {
458    let binary_hash = std::env::current_exe()
459        .ok()
460        .and_then(|p| hash_file(&p).ok());
461    Json(VersionInfo {
462        version: env!("CARGO_PKG_VERSION"),
463        name: env!("CARGO_PKG_NAME"),
464        binary_hash,
465    })
466}
467
468/// List sessions
469#[derive(Deserialize)]
470struct ListSessionsQuery {
471    limit: Option<usize>,
472}
473
474async fn list_sessions(
475    Query(query): Query<ListSessionsQuery>,
476) -> Result<Json<Vec<crate::session::SessionSummary>>, (StatusCode, String)> {
477    let sessions = crate::session::list_sessions()
478        .await
479        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
480
481    let limit = query.limit.unwrap_or(50);
482    Ok(Json(sessions.into_iter().take(limit).collect()))
483}
484
485/// Create a new session
486#[derive(Deserialize)]
487struct CreateSessionRequest {
488    title: Option<String>,
489    agent: Option<String>,
490}
491
492async fn create_session(
493    Json(req): Json<CreateSessionRequest>,
494) -> Result<Json<crate::session::Session>, (StatusCode, String)> {
495    let mut session = crate::session::Session::new()
496        .await
497        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
498
499    session.title = req.title;
500    if let Some(agent) = req.agent {
501        session.agent = agent;
502    }
503
504    session
505        .save()
506        .await
507        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
508
509    Ok(Json(session))
510}
511
512/// Get a session by ID
513async fn get_session(
514    axum::extract::Path(id): axum::extract::Path<String>,
515) -> Result<Json<crate::session::Session>, (StatusCode, String)> {
516    let session = crate::session::Session::load(&id)
517        .await
518        .map_err(|e| (StatusCode::NOT_FOUND, e.to_string()))?;
519
520    Ok(Json(session))
521}
522
523/// Prompt a session
524#[derive(Deserialize)]
525struct PromptRequest {
526    message: String,
527}
528
529async fn prompt_session(
530    axum::extract::Path(id): axum::extract::Path<String>,
531    Json(req): Json<PromptRequest>,
532) -> Result<Json<crate::session::SessionResult>, (StatusCode, String)> {
533    // Validate the message is not empty
534    if req.message.trim().is_empty() {
535        return Err((
536            StatusCode::BAD_REQUEST,
537            "Message cannot be empty".to_string(),
538        ));
539    }
540
541    // Log the prompt request (uses the message field)
542    tracing::info!(
543        session_id = %id,
544        message_len = req.message.len(),
545        "Received prompt request"
546    );
547
548    // TODO: Implement actual prompting
549    Err((
550        StatusCode::NOT_IMPLEMENTED,
551        "Prompt execution not yet implemented".to_string(),
552    ))
553}
554
555/// Get configuration
556async fn get_config(State(state): State<AppState>) -> Json<Config> {
557    Json((*state.config).clone())
558}
559
560/// List providers
561async fn list_providers() -> Json<Vec<String>> {
562    Json(vec![
563        "openai".to_string(),
564        "anthropic".to_string(),
565        "google".to_string(),
566    ])
567}
568
569/// List agents
570async fn list_agents() -> Json<Vec<crate::agent::AgentInfo>> {
571    let registry = crate::agent::AgentRegistry::with_builtins();
572    Json(registry.list().into_iter().cloned().collect())
573}
574
575async fn start_cognition(
576    State(state): State<AppState>,
577    payload: Option<Json<StartCognitionRequest>>,
578) -> Result<Json<CognitionStatus>, (StatusCode, String)> {
579    state
580        .cognition
581        .start(payload.map(|Json(body)| body))
582        .await
583        .map(Json)
584        .map_err(internal_error)
585}
586
587async fn stop_cognition(
588    State(state): State<AppState>,
589    payload: Option<Json<StopCognitionRequest>>,
590) -> Result<Json<CognitionStatus>, (StatusCode, String)> {
591    let reason = payload.and_then(|Json(body)| body.reason);
592    state
593        .cognition
594        .stop(reason)
595        .await
596        .map(Json)
597        .map_err(internal_error)
598}
599
600async fn get_cognition_status(
601    State(state): State<AppState>,
602) -> Result<Json<CognitionStatus>, (StatusCode, String)> {
603    Ok(Json(state.cognition.status().await))
604}
605
606async fn stream_cognition(
607    State(state): State<AppState>,
608) -> Sse<impl futures::Stream<Item = Result<Event, Infallible>>> {
609    let rx = state.cognition.subscribe_events();
610
611    let event_stream = stream::unfold(rx, |mut rx| async move {
612        match rx.recv().await {
613            Ok(event) => {
614                let payload = serde_json::to_string(&event).unwrap_or_else(|_| "{}".to_string());
615                let sse_event = Event::default().event("cognition").data(payload);
616                Some((Ok(sse_event), rx))
617            }
618            Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
619                let lag_event = Event::default()
620                    .event("lag")
621                    .data(format!("skipped {}", skipped));
622                Some((Ok(lag_event), rx))
623            }
624            Err(tokio::sync::broadcast::error::RecvError::Closed) => None,
625        }
626    });
627
628    Sse::new(event_stream).keep_alive(KeepAlive::new().interval(std::time::Duration::from_secs(15)))
629}
630
631async fn get_latest_snapshot(
632    State(state): State<AppState>,
633) -> Result<Json<MemorySnapshot>, (StatusCode, String)> {
634    match state.cognition.latest_snapshot().await {
635        Some(snapshot) => Ok(Json(snapshot)),
636        None => Err((StatusCode::NOT_FOUND, "No snapshots available".to_string())),
637    }
638}
639
640async fn create_persona(
641    State(state): State<AppState>,
642    Json(req): Json<CreatePersonaRequest>,
643) -> Result<Json<crate::cognition::PersonaRuntimeState>, (StatusCode, String)> {
644    state
645        .cognition
646        .create_persona(req)
647        .await
648        .map(Json)
649        .map_err(internal_error)
650}
651
652async fn spawn_persona(
653    State(state): State<AppState>,
654    Path(id): Path<String>,
655    Json(req): Json<SpawnPersonaRequest>,
656) -> Result<Json<crate::cognition::PersonaRuntimeState>, (StatusCode, String)> {
657    state
658        .cognition
659        .spawn_child(&id, req)
660        .await
661        .map(Json)
662        .map_err(internal_error)
663}
664
665async fn reap_persona(
666    State(state): State<AppState>,
667    Path(id): Path<String>,
668    payload: Option<Json<ReapPersonaRequest>>,
669) -> Result<Json<ReapPersonaResponse>, (StatusCode, String)> {
670    let req = payload
671        .map(|Json(body)| body)
672        .unwrap_or(ReapPersonaRequest {
673            cascade: Some(false),
674            reason: None,
675        });
676
677    state
678        .cognition
679        .reap_persona(&id, req)
680        .await
681        .map(Json)
682        .map_err(internal_error)
683}
684
685async fn get_swarm_lineage(
686    State(state): State<AppState>,
687) -> Result<Json<LineageGraph>, (StatusCode, String)> {
688    Ok(Json(state.cognition.lineage_graph().await))
689}
690
691// ── Belief, Attention, Governance, Workspace handlers ──
692
693#[derive(Deserialize)]
694struct BeliefFilter {
695    status: Option<String>,
696    persona: Option<String>,
697}
698
699async fn list_beliefs(
700    State(state): State<AppState>,
701    Query(filter): Query<BeliefFilter>,
702) -> Result<Json<Vec<Belief>>, (StatusCode, String)> {
703    let beliefs = state.cognition.get_beliefs().await;
704    let mut result: Vec<Belief> = beliefs.into_values().collect();
705
706    if let Some(status) = &filter.status {
707        result.retain(|b| {
708            let s = serde_json::to_string(&b.status).unwrap_or_default();
709            s.contains(status)
710        });
711    }
712    if let Some(persona) = &filter.persona {
713        result.retain(|b| &b.asserted_by == persona);
714    }
715
716    result.sort_by(|a, b| {
717        b.confidence
718            .partial_cmp(&a.confidence)
719            .unwrap_or(std::cmp::Ordering::Equal)
720    });
721    Ok(Json(result))
722}
723
724async fn get_belief(
725    State(state): State<AppState>,
726    Path(id): Path<String>,
727) -> Result<Json<Belief>, (StatusCode, String)> {
728    match state.cognition.get_belief(&id).await {
729        Some(belief) => Ok(Json(belief)),
730        None => Err((StatusCode::NOT_FOUND, format!("Belief not found: {}", id))),
731    }
732}
733
734async fn list_attention(
735    State(state): State<AppState>,
736) -> Result<Json<Vec<AttentionItem>>, (StatusCode, String)> {
737    Ok(Json(state.cognition.get_attention_queue().await))
738}
739
740async fn list_proposals(
741    State(state): State<AppState>,
742) -> Result<Json<Vec<Proposal>>, (StatusCode, String)> {
743    let proposals = state.cognition.get_proposals().await;
744    let mut result: Vec<Proposal> = proposals.into_values().collect();
745    result.sort_by(|a, b| b.created_at.cmp(&a.created_at));
746    Ok(Json(result))
747}
748
749async fn approve_proposal(
750    State(state): State<AppState>,
751    Path(id): Path<String>,
752) -> Result<Json<serde_json::Value>, (StatusCode, String)> {
753    state
754        .cognition
755        .approve_proposal(&id)
756        .await
757        .map(|_| Json(serde_json::json!({ "approved": true, "proposal_id": id })))
758        .map_err(internal_error)
759}
760
761async fn list_receipts(
762    State(state): State<AppState>,
763) -> Result<Json<Vec<DecisionReceipt>>, (StatusCode, String)> {
764    Ok(Json(state.cognition.get_receipts().await))
765}
766
767async fn get_workspace(
768    State(state): State<AppState>,
769) -> Result<Json<GlobalWorkspace>, (StatusCode, String)> {
770    Ok(Json(state.cognition.get_workspace().await))
771}
772
773async fn get_governance(
774    State(state): State<AppState>,
775) -> Result<Json<crate::cognition::SwarmGovernance>, (StatusCode, String)> {
776    Ok(Json(state.cognition.get_governance().await))
777}
778
779async fn get_persona(
780    State(state): State<AppState>,
781    axum::extract::Path(id): axum::extract::Path<String>,
782) -> Result<Json<crate::cognition::PersonaRuntimeState>, (StatusCode, String)> {
783    state
784        .cognition
785        .get_persona(&id)
786        .await
787        .map(Json)
788        .ok_or_else(|| (StatusCode::NOT_FOUND, format!("Persona not found: {}", id)))
789}
790
791// ── Audit trail endpoints ──
792
793#[derive(Deserialize)]
794struct AuditQuery {
795    limit: Option<usize>,
796    category: Option<String>,
797}
798
799async fn list_audit_entries(
800    State(state): State<AppState>,
801    Query(query): Query<AuditQuery>,
802) -> Result<Json<Vec<audit::AuditEntry>>, (StatusCode, String)> {
803    let limit = query.limit.unwrap_or(100).min(1000);
804
805    let entries = if let Some(ref cat) = query.category {
806        let category = match cat.as_str() {
807            "api" => AuditCategory::Api,
808            "tool" | "tool_execution" => AuditCategory::ToolExecution,
809            "session" => AuditCategory::Session,
810            "cognition" => AuditCategory::Cognition,
811            "swarm" => AuditCategory::Swarm,
812            "auth" => AuditCategory::Auth,
813            "k8s" => AuditCategory::K8s,
814            "sandbox" => AuditCategory::Sandbox,
815            "config" => AuditCategory::Config,
816            _ => {
817                return Err((
818                    StatusCode::BAD_REQUEST,
819                    format!("Unknown category: {}", cat),
820                ));
821            }
822        };
823        state.audit_log.by_category(category, limit).await
824    } else {
825        state.audit_log.recent(limit).await
826    };
827
828    Ok(Json(entries))
829}
830
831// ── K8s self-deployment endpoints ──
832
833async fn get_k8s_status(
834    State(state): State<AppState>,
835) -> Result<Json<crate::k8s::K8sStatus>, (StatusCode, String)> {
836    Ok(Json(state.k8s.status().await))
837}
838
839#[derive(Deserialize)]
840struct ScaleRequest {
841    replicas: i32,
842}
843
844async fn k8s_scale(
845    State(state): State<AppState>,
846    Json(req): Json<ScaleRequest>,
847) -> Result<Json<crate::k8s::DeployAction>, (StatusCode, String)> {
848    if req.replicas < 0 || req.replicas > 100 {
849        return Err((
850            StatusCode::BAD_REQUEST,
851            "Replicas must be between 0 and 100".to_string(),
852        ));
853    }
854
855    state
856        .audit_log
857        .log(
858            AuditCategory::K8s,
859            format!("scale:{}", req.replicas),
860            AuditOutcome::Success,
861            None,
862            None,
863        )
864        .await;
865
866    state
867        .k8s
868        .scale(req.replicas)
869        .await
870        .map(Json)
871        .map_err(internal_error)
872}
873
874async fn k8s_restart(
875    State(state): State<AppState>,
876) -> Result<Json<crate::k8s::DeployAction>, (StatusCode, String)> {
877    state
878        .audit_log
879        .log(
880            AuditCategory::K8s,
881            "rolling_restart",
882            AuditOutcome::Success,
883            None,
884            None,
885        )
886        .await;
887
888    state
889        .k8s
890        .rolling_restart()
891        .await
892        .map(Json)
893        .map_err(internal_error)
894}
895
896async fn k8s_list_pods(
897    State(state): State<AppState>,
898) -> Result<Json<Vec<crate::k8s::PodInfo>>, (StatusCode, String)> {
899    state
900        .k8s
901        .list_pods()
902        .await
903        .map(Json)
904        .map_err(internal_error)
905}
906
907async fn k8s_actions(
908    State(state): State<AppState>,
909) -> Result<Json<Vec<crate::k8s::DeployAction>>, (StatusCode, String)> {
910    Ok(Json(state.k8s.recent_actions(100).await))
911}
912
913#[derive(Deserialize)]
914struct SpawnSubagentRequest {
915    subagent_id: String,
916    #[serde(default)]
917    image: Option<String>,
918    #[serde(default)]
919    env_vars: std::collections::HashMap<String, String>,
920}
921
922async fn k8s_spawn_subagent(
923    State(state): State<AppState>,
924    Json(req): Json<SpawnSubagentRequest>,
925) -> Result<Json<crate::k8s::DeployAction>, (StatusCode, String)> {
926    state
927        .k8s
928        .spawn_subagent_pod(&req.subagent_id, req.image.as_deref(), req.env_vars)
929        .await
930        .map(Json)
931        .map_err(internal_error)
932}
933
934async fn k8s_delete_subagent(
935    State(state): State<AppState>,
936    axum::extract::Path(id): axum::extract::Path<String>,
937) -> Result<Json<crate::k8s::DeployAction>, (StatusCode, String)> {
938    state
939        .k8s
940        .delete_subagent_pod(&id)
941        .await
942        .map(Json)
943        .map_err(internal_error)
944}
945
946/// List registered plugins.
947async fn list_plugins(State(_state): State<AppState>) -> Json<PluginListResponse> {
948    let server_fingerprint = hash_bytes(env!("CARGO_PKG_VERSION").as_bytes());
949    let signing_key = SigningKey::from_env();
950    let test_sig = signing_key.sign("_probe", "0.0.0", &server_fingerprint);
951    Json(PluginListResponse {
952        server_fingerprint,
953        signing_available: !test_sig.is_empty(),
954        plugins: Vec::<PluginManifest>::new(),
955    })
956}
957
958#[derive(Serialize)]
959struct PluginListResponse {
960    server_fingerprint: String,
961    signing_available: bool,
962    plugins: Vec<PluginManifest>,
963}
964
965fn internal_error(error: anyhow::Error) -> (StatusCode, String) {
966    let message = error.to_string();
967    if message.contains("not found") {
968        return (StatusCode::NOT_FOUND, message);
969    }
970    if message.contains("disabled") || message.contains("exceeds") || message.contains("limit") {
971        return (StatusCode::BAD_REQUEST, message);
972    }
973    (StatusCode::INTERNAL_SERVER_ERROR, message)
974}
975
976fn env_bool(name: &str, default: bool) -> bool {
977    std::env::var(name)
978        .ok()
979        .and_then(|v| match v.to_ascii_lowercase().as_str() {
980            "1" | "true" | "yes" | "on" => Some(true),
981            "0" | "false" | "no" | "off" => Some(false),
982            _ => None,
983        })
984        .unwrap_or(default)
985}
codetether_agent/server/mod.rs

codetether_agent/server/
mod.rs
