oikos/internal/httpapi/activity.go

package httpapi

import (
	"encoding/json"
	"log/slog"
	"net/http"
	"strconv"
	"strings"

	"github.com/go-chi/chi/v5"
)

// activityItem is one row in the global activity feed — a human-readable
// projection of an execution, independent of the paginated/alphabetically-
// sorted ListExecutions (which orders by target slug for entity-scoped
// browsing, not recency — wrong shape for "what just happened").
type activityItem struct {
	ID          string  `json:"id"`
	Target      string  `json:"target"`
	Verb        string  `json:"verb"`    // e.g. "run", "pct_create", "systemctl"
	Summary     string  `json:"summary"` // human-readable: the command, or purpose, or action detail
	RiskClass   string  `json:"risk_class"`
	Status      string  `json:"status"`
	DurationMs  *int    `json:"duration_ms"`
	Error       string  `json:"error,omitempty"`
	CreatedAt   string  `json:"created_at"`
	CompletedAt *string `json:"completed_at"`
}

// splitAction parses the "verb:params" encoding used throughout executions.action
// (see internal/mcp/server.go) into a verb and a human-readable summary. For
// `run`, params is JSON {command, purpose} — show the purpose if present
// (it's written for a human), falling back to the raw command. For other
// actions (pct_create, systemctl, apt_upgrade, pct_exec), params is either a
// JSON blob or a short flag string — truncate either as a fallback summary.
func splitAction(action string) (verb, summary string) {
	idx := strings.IndexByte(action, ':')
	if idx < 0 {
		return action, ""
	}
	verb, params := action[:idx], action[idx+1:]
	if verb == "run" {
		var p struct {
			Command string `json:"command"`
			Purpose string `json:"purpose"`
		}
		if json.Unmarshal([]byte(params), &p) == nil {
			if p.Purpose != "" {
				return verb, p.Purpose
			}
			return verb, p.Command
		}
	}
	if verb == "pct_create" {
		var p struct {
			Hostname string `json:"hostname"`
		}
		if json.Unmarshal([]byte(params), &p) == nil && p.Hostname != "" {
			return verb, "provision " + p.Hostname
		}
	}
	if len(params) > 140 {
		params = params[:140] + "…"
	}
	return verb, params
}

// serveRecentActivity backs the Operations page's live activity feed — the
// global "what is the system doing / what did it just do" view, recency-
// ordered (unlike ListExecutions, which sorts by target for pagination).
// Custom route, same shape/rationale as serveRecentKnowledge.
func (s *Server) serveRecentActivity(w http.ResponseWriter, req *http.Request) {
	ctx := req.Context()
	limit := 50
	if l := req.URL.Query().Get("limit"); l != "" {
		if n, err := strconv.Atoi(l); err == nil && n > 0 && n <= 200 {
			limit = n
		}
	}

	rows, err := s.pool.Query(ctx, `
		SELECT e.entity_id, te.slug, e.action, e.risk_class, e.status,
		       e.duration_ms, e.result, e.created_at::text, e.completed_at::text
		FROM executions e
		JOIN entities te ON te.id = e.target_entity_id
		ORDER BY e.created_at DESC
		LIMIT $1`, limit)
	if err != nil {
		writeProblem(w, req, http.StatusInternalServerError, "query failed", err.Error())
		return
	}
	defer rows.Close()

	items := []activityItem{}
	for rows.Next() {
		var it activityItem
		var action string
		var resultBytes []byte
		var completedAt *string
		if err := rows.Scan(&it.ID, &it.Target, &action, &it.RiskClass, &it.Status,
			&it.DurationMs, &resultBytes, &it.CreatedAt, &completedAt); err != nil {
			slog.Error("httpapi: activity/recent row scan failed", "error", err)
			continue
		}
		it.Verb, it.Summary = splitAction(action)
		it.CompletedAt = completedAt
		if len(resultBytes) > 0 {
			var result map[string]any
			if json.Unmarshal(resultBytes, &result) == nil {
				if e, ok := result["error"].(string); ok && e != "" {
					if len(e) > 200 {
						e = e[:200] + "…"
					}
					it.Error = e
				}
			}
		}
		items = append(items, it)
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(map[string]any{"items": items})
}

// sessionDigestItem summarizes one execution for the session digest.
type sessionDigestItem struct {
	Target    string `json:"target"`
	Verb      string `json:"verb"`
	Summary   string `json:"summary"`
	RiskClass string `json:"risk_class"`
	Status    string `json:"status"`
}

// serveSessionDigest answers "what did THIS chat session actually do" —
// commands run (grouped by outcome), distinct entities touched, and knowledge
// written during the session's time window. Uses nomos_plan_executions (the
// session<->execution link added for auto-continuation) as the source of
// truth for which executions belong to this session; knowledge correlation is
// a best-effort time-window match since knowledge_entities has no session_id.
func (s *Server) serveSessionDigest(w http.ResponseWriter, req *http.Request) {
	ctx := req.Context()
	sessionID := chi.URLParam(req, "id")
	if sessionID == "" {
		writeProblem(w, req, http.StatusBadRequest, "missing session id", "")
		return
	}

	rows, err := s.pool.Query(ctx, `
		SELECT te.slug, e.action, e.risk_class, e.status
		FROM nomos_plan_executions l
		JOIN executions e ON e.entity_id = l.execution_id
		JOIN entities te ON te.id = e.target_entity_id
		WHERE l.session_id = $1
		ORDER BY e.created_at`, sessionID)
	if err != nil {
		writeProblem(w, req, http.StatusInternalServerError, "query failed", err.Error())
		return
	}
	defer rows.Close()

	items := []sessionDigestItem{}
	byStatus := map[string]int{}
	targets := map[string]bool{}
	for rows.Next() {
		var it sessionDigestItem
		var action string
		if err := rows.Scan(&it.Target, &action, &it.RiskClass, &it.Status); err != nil {
			continue
		}
		it.Verb, it.Summary = splitAction(action)
		items = append(items, it)
		byStatus[it.Status]++
		targets[it.Target] = true
	}

	entityList := make([]string, 0, len(targets))
	for t := range targets {
		entityList = append(entityList, t)
	}

	// Best-effort knowledge correlation: notes the agent wrote during this
	// session's active window. Not exact (no session_id on knowledge_entities)
	// but close enough to show "you learned N things in this session".
	var knowledgeTitles []string
	krows, err := s.pool.Query(ctx, `
		SELECT ke.title FROM knowledge_entities ke
		WHERE ke.source = 'nomos-agent'
		  AND ke.updated_at BETWEEN
		    (SELECT COALESCE(MIN(created_at), now()) FROM agent_messages WHERE session_id = $1)
		    AND
		    (SELECT COALESCE(MAX(created_at), now()) + interval '2 minutes' FROM agent_messages WHERE session_id = $1)
		ORDER BY ke.updated_at`, sessionID)
	if err == nil {
		defer krows.Close()
		for krows.Next() {
			var t string
			if krows.Scan(&t) == nil {
				knowledgeTitles = append(knowledgeTitles, t)
			}
		}
	}
	if knowledgeTitles == nil {
		knowledgeTitles = []string{}
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(map[string]any{
		"session_id":        sessionID,
		"total_executions":  len(items),
		"by_status":         byStatus,
		"entities_touched":  entityList,
		"executions":        items,
		"knowledge_created": knowledgeTitles,
	})
}