feat(sessions): add SessionAdapter — REST CRUD for sessions/milestones/episodes (#50)

New internal/sessions package with Handler bridging 7 REST endpoints to: - POST /memory/ingest on poimen-memory:8080 (writes) - GraphQL hybridSearch/entity on poimen-graphql:8090 (reads) Adapter registration: X-Service: sessions with resources sessions, session, milestones, episodes, episode. IDs via query params (?session_id=, ?episode_id=) — dispatcher does not substitute path params. episode_id = sha256(session_id+milestone_id+attempt) per §4.14 Q6. Content JSON-encoded (prevents key:value injection on user fields). Limit capped at 200. GraphQL error body detected and surfaced as 502. tracing.NewTransport wired for cross-service trace propagation. JWT forwarded to both upstreams for auth. 30 tests in handler_test.go covering all 7 handlers + EpisodeID helper. Closes #50
2026-09-28 09:39:19 +08:00
parent 0bde5dd8fe
commit e8ee03c8fc
5 changed files with 1214 additions and 0 deletions
@@ -0,0 +1,476 @@
+// Package sessions provides the internal handler for the sessions service adapter.
+// It bridges the api-gw REST surface to:
+//   - poimen-memory  /memory/ingest   (writes — SESSION, MILESTONE, EPISODE entities)
+//   - poimen-graphql /graphql          (reads  — hybridSearch, entity queries)
+//
+// All IDs passed by the caller (session_id, episode_id) go as query params because
+// the ServiceAdapter dispatcher does not substitute URL path parameters.
+// episode_id is sha256(session_id + milestone_id + attempt) — callers compute it the
+// same way; the handler accepts it on write and returns it on read.
+package sessions
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/google/uuid"
+
+	"forgejo.riotpiao.com/rock/homelab-frontend/internal/tracing"
+)
+
+const (
+	defaultMemoryURL  = "http://poimen-memory.poimen.svc.cluster.local:8080"
+	defaultGraphQLURL = "http://poimen-graphql.poimen.svc.cluster.local:8090/graphql"
+)
+
+// Handler handles session REST endpoints via internal service calls.
+type Handler struct {
+	memoryURL  string
+	graphqlURL string
+	client     *http.Client
+}
+
+// NewHandler creates a Handler reading upstream URLs from env vars.
+// MEMORY_UPSTREAM_URL  — default: defaultMemoryURL
+// GRAPHQL_UPSTREAM_URL — default: defaultGraphQLURL
+func NewHandler() *Handler {
+	mem := os.Getenv("MEMORY_UPSTREAM_URL")
+	if mem == "" {
+		mem = defaultMemoryURL
+	}
+	gql := os.Getenv("GRAPHQL_UPSTREAM_URL")
+	if gql == "" {
+		gql = defaultGraphQLURL
+	}
+	return &Handler{
+		memoryURL:  mem,
+		graphqlURL: gql,
+		client:     &http.Client{Timeout: 30 * time.Second, Transport: tracing.NewTransport(nil)},
+	}
+}
+
+// ServeHTTP routes by X-Upstream-Path set by the dispatcher.
+func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	path := r.Header.Get("X-Upstream-Path")
+	switch path {
+	case "/v1/sessions/create":
+		h.handleCreate(w, r)
+	case "/v1/sessions/list":
+		h.handleList(w, r)
+	case "/v1/sessions/get":
+		h.handleGet(w, r)
+	case "/v1/sessions/patch":
+		h.handlePatch(w, r)
+	case "/v1/sessions/milestones/create":
+		h.handleCreateMilestone(w, r)
+	case "/v1/sessions/episodes/list":
+		h.handleListEpisodes(w, r)
+	case "/v1/sessions/episodes/get":
+		h.handleGetEpisode(w, r)
+	default:
+		writeJSON(w, http.StatusNotFound, map[string]string{"error": "sessions endpoint not found"})
+	}
+}
+
+// ── request / response types ──────────────────────────────────────────────────
+
+type createSessionReq struct {
+	Name string `json:"name"`
+	Goal string `json:"goal,omitempty"`
+	Cwd  string `json:"cwd,omitempty"`
+}
+
+type sessionResp struct {
+	SessionID     string `json:"session_id"`
+	Name          string `json:"name"`
+	Goal          string `json:"goal,omitempty"`
+	Status        string `json:"status"`
+	MemoryProject string `json:"memory_project"`
+	Cwd           string `json:"cwd,omitempty"`
+	CreatedAt     string `json:"created_at"`
+}
+
+type patchSessionReq struct {
+	Name   *string `json:"name,omitempty"`
+	Goal   *string `json:"goal,omitempty"`
+	Status *string `json:"status,omitempty"`
+}
+
+type createMilestoneReq struct {
+	Title    string  `json:"title"`
+	OrderIdx int     `json:"order_idx"`
+	ParentID *string `json:"parent_id,omitempty"`
+}
+
+type milestoneResp struct {
+	MilestoneID string  `json:"milestone_id"`
+	SessionID   string  `json:"session_id"`
+	Title       string  `json:"title"`
+	Status      string  `json:"status"`
+	OrderIdx    int     `json:"order_idx"`
+	ParentID    *string `json:"parent_id,omitempty"`
+	CreatedAt   string  `json:"created_at"`
+}
+
+// ── handlers ──────────────────────────────────────────────────────────────────
+
+// POST /v1/sessions  (X-Resource: sessions)
+func (h *Handler) handleCreate(w http.ResponseWriter, r *http.Request) {
+	var req createSessionReq
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
+		return
+	}
+	if req.Name == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "name is required"})
+		return
+	}
+
+	sessionID := uuid.New().String()
+	project := "session-" + sessionID
+	now := time.Now().UTC().Format(time.RFC3339)
+
+	contentBytes, _ := json.Marshal(map[string]any{
+		"type":           "SESSION",
+		"id":             sessionID,
+		"name":           req.Name,
+		"goal":           req.Goal,
+		"status":         "active",
+		"cwd":            req.Cwd,
+		"memory_project": project,
+		"created_at":     now,
+	})
+	content := string(contentBytes)
+	if err := h.ingest(r, project, sessionID, content); err != nil {
+		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "ingest failed: " + err.Error()})
+		return
+	}
+
+	writeJSON(w, http.StatusCreated, sessionResp{
+		SessionID:     sessionID,
+		Name:          req.Name,
+		Goal:          req.Goal,
+		Status:        "active",
+		MemoryProject: project,
+		Cwd:           req.Cwd,
+		CreatedAt:     now,
+	})
+}
+
+// GET /v1/sessions  (X-Resource: sessions)  ?status=active&cwd=/path
+func (h *Handler) handleList(w http.ResponseWriter, r *http.Request) {
+	q := r.URL.Query()
+	status := q.Get("status")
+	cwd := q.Get("cwd")
+	limitStr := q.Get("limit")
+	limit := 50
+	if limitStr != "" {
+		fmt.Sscanf(limitStr, "%d", &limit) //nolint:errcheck
+	}
+	if limit > 200 {
+		limit = 200
+	}
+
+	searchQuery := "session"
+	if status != "" {
+		searchQuery = fmt.Sprintf("session status:%s", status)
+	}
+	if cwd != "" {
+		searchQuery = fmt.Sprintf("%s cwd:%s", searchQuery, cwd)
+	}
+
+	gqlQuery := `query($q:String!,$limit:Int!){
+		hybridSearch(query:$q,limit:$limit,entityType:SESSION){
+			entities{id name summary tCreated}
+			searchMode
+		}
+	}`
+	vars := map[string]any{"q": searchQuery, "limit": limit}
+	result, err := h.gql(r, gqlQuery, vars)
+	if err != nil {
+		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "graphql query failed: " + err.Error()})
+		return
+	}
+	writeJSON(w, http.StatusOK, result)
+}
+
+// GET /v1/sessions/:id  (X-Resource: session)  ?session_id=<uuid>
+func (h *Handler) handleGet(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.URL.Query().Get("session_id")
+	if sessionID == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"})
+		return
+	}
+
+	project := "session-" + sessionID
+	gqlQuery := `query($id:String!,$project:String!){
+		hybridSearch(query:$id,project:$project,limit:1,entityType:SESSION){
+			entities{id name summary tCreated}
+		}
+	}`
+	vars := map[string]any{"id": sessionID, "project": project}
+	result, err := h.gql(r, gqlQuery, vars)
+	if err != nil {
+		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "graphql query failed: " + err.Error()})
+		return
+	}
+	writeJSON(w, http.StatusOK, result)
+}
+
+// PATCH /v1/sessions/:id  (X-Resource: session)  ?session_id=<uuid>
+func (h *Handler) handlePatch(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.URL.Query().Get("session_id")
+	if sessionID == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"})
+		return
+	}
+
+	var req patchSessionReq
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
+		return
+	}
+
+	now := time.Now().UTC().Format(time.RFC3339)
+	patch := map[string]any{
+		"type":       "SESSION",
+		"id":         sessionID,
+		"updated_at": now,
+	}
+	if req.Name != nil {
+		patch["name"] = *req.Name
+	}
+	if req.Goal != nil {
+		patch["goal"] = *req.Goal
+	}
+	if req.Status != nil {
+		patch["status"] = *req.Status
+		if *req.Status == "archived" {
+			patch["archived_at"] = now
+		}
+	}
+	contentBytes, _ := json.Marshal(patch)
+
+	project := "session-" + sessionID
+	if err := h.ingest(r, project, sessionID+"-patch", string(contentBytes)); err != nil {
+		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "ingest failed: " + err.Error()})
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]string{"session_id": sessionID, "status": "updated"})
+}
+
+// POST /v1/sessions/:id/milestones  (X-Resource: milestones)  ?session_id=<uuid>
+func (h *Handler) handleCreateMilestone(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.URL.Query().Get("session_id")
+	if sessionID == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"})
+		return
+	}
+
+	var req createMilestoneReq
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
+		return
+	}
+	if req.Title == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "title is required"})
+		return
+	}
+
+	milestoneID := uuid.New().String()
+	now := time.Now().UTC().Format(time.RFC3339)
+	project := "session-" + sessionID
+
+	msContent := map[string]any{
+		"type":       "MILESTONE",
+		"id":         milestoneID,
+		"session_id": sessionID,
+		"title":      req.Title,
+		"status":     "pending",
+		"order_idx":  req.OrderIdx,
+		"created_at": now,
+	}
+	if req.ParentID != nil {
+		msContent["parent_id"] = *req.ParentID
+	}
+	contentBytes, _ := json.Marshal(msContent)
+	content := string(contentBytes)
+	if err := h.ingest(r, project, milestoneID, content); err != nil {
+		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "ingest failed: " + err.Error()})
+		return
+	}
+
+	writeJSON(w, http.StatusCreated, milestoneResp{
+		MilestoneID: milestoneID,
+		SessionID:   sessionID,
+		Title:       req.Title,
+		Status:      "pending",
+		OrderIdx:    req.OrderIdx,
+		ParentID:    req.ParentID,
+		CreatedAt:   now,
+	})
+}
+
+// GET /v1/sessions/:id/episodes  (X-Resource: episodes)  ?session_id=<uuid>
+func (h *Handler) handleListEpisodes(w http.ResponseWriter, r *http.Request) {
+	sessionID := r.URL.Query().Get("session_id")
+	if sessionID == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "session_id query param required"})
+		return
+	}
+
+	limitStr := r.URL.Query().Get("limit")
+	limit := 50
+	if limitStr != "" {
+		fmt.Sscanf(limitStr, "%d", &limit) //nolint:errcheck
+	}
+	if limit > 200 {
+		limit = 200
+	}
+
+	project := "session-" + sessionID
+	gqlQuery := `query($project:String!,$limit:Int!){
+		hybridSearch(query:"episode",project:$project,limit:$limit,entityType:EPISODE){
+			entities{id name summary tCreated}
+			searchMode
+		}
+	}`
+	vars := map[string]any{"project": project, "limit": limit}
+	result, err := h.gql(r, gqlQuery, vars)
+	if err != nil {
+		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "graphql query failed: " + err.Error()})
+		return
+	}
+	writeJSON(w, http.StatusOK, result)
+}
+
+// GET /v1/sessions/:id/episodes/:eid  (X-Resource: episode)
+// ?session_id=<uuid>&episode_id=<sha256>
+func (h *Handler) handleGetEpisode(w http.ResponseWriter, r *http.Request) {
+	q := r.URL.Query()
+	sessionID := q.Get("session_id")
+	episodeID := q.Get("episode_id")
+	if sessionID == "" || episodeID == "" {
+		writeJSON(w, http.StatusBadRequest, map[string]string{
+			"error": "session_id and episode_id query params required",
+		})
+		return
+	}
+
+	gqlQuery := `query($id:ID!){
+		entity(id:$id){
+			id name entityType summary tCreated
+			edges{
+				relationType fact confidence
+				target{name entityType}
+			}
+		}
+	}`
+	vars := map[string]any{"id": episodeID}
+	result, err := h.gql(r, gqlQuery, vars)
+	if err != nil {
+		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "graphql query failed: " + err.Error()})
+		return
+	}
+	writeJSON(w, http.StatusOK, result)
+}
+
+// ── helpers ───────────────────────────────────────────────────────────────────
+
+// ingest calls POST /memory/ingest on the memory service, forwarding the caller's JWT.
+func (h *Handler) ingest(r *http.Request, project, recordID, content string) error {
+	body, err := json.Marshal(map[string]any{
+		"project": project,
+		"source":  "sessions-api",
+		"records": []map[string]string{
+			{"id": recordID, "content": content},
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("marshal: %w", err)
+	}
+
+	req, err := http.NewRequestWithContext(r.Context(), http.MethodPost,
+		h.memoryURL+"/memory/ingest", bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("build request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	if auth := r.Header.Get("Authorization"); auth != "" {
+		req.Header.Set("Authorization", auth)
+	}
+
+	resp, err := h.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("http: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 300 {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+		return fmt.Errorf("memory ingest returned %d: %s", resp.StatusCode, body)
+	}
+	return nil
+}
+
+// gql executes a GraphQL query against poimen-graphql, forwarding the caller's JWT.
+func (h *Handler) gql(r *http.Request, query string, variables map[string]any) (any, error) {
+	payload, err := json.Marshal(map[string]any{
+		"query":     query,
+		"variables": variables,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("marshal: %w", err)
+	}
+
+	req, err := http.NewRequestWithContext(r.Context(), http.MethodPost,
+		h.graphqlURL, bytes.NewReader(payload))
+	if err != nil {
+		return nil, fmt.Errorf("build request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	if auth := r.Header.Get("Authorization"); auth != "" {
+		req.Header.Set("Authorization", auth)
+	}
+
+	resp, err := h.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("http: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 300 {
+		return nil, fmt.Errorf("graphql returned %d", resp.StatusCode)
+	}
+
+	var result map[string]any
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return nil, fmt.Errorf("decode: %w", err)
+	}
+	if errs, ok := result["errors"]; ok && errs != nil {
+		b, _ := json.Marshal(errs)
+		return nil, fmt.Errorf("graphql errors: %s", b)
+	}
+	return result, nil
+}
+
+// EpisodeID computes the deterministic episode ID per §4.14 Q6:
+// sha256(session_id + milestone_id + attempt).
+// Exported so callers can verify the value the server would produce.
+func EpisodeID(sessionID, milestoneID string, attempt int) string {
+	h := sha256.New()
+	h.Write([]byte(fmt.Sprintf("%s%s%d", sessionID, milestoneID, attempt)))
+	return fmt.Sprintf("%x", h.Sum(nil))
+}
+
+func writeJSON(w http.ResponseWriter, status int, data any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	json.NewEncoder(w).Encode(data) //nolint:errcheck
+}