Admin Bot
bb7a490f5c
CI / CI (pull_request) Successful in 2m10s
hybridSearch mandates a project scope, but sessions scatter across per-session projects (session-<uuid>). The entities query accepts optional project — omit it to list SESSION entities across all projects. Also adds projectId to the list response so callers can scope follow-up queries to the right session project. Fixes: GET X-Service:sessions X-Resource:sessions returning 502
515 lines
16 KiB
Go
515 lines
16 KiB
Go
// Package sessions provides the internal handler for the sessions service adapter.
|
|
// It bridges the api-gw REST surface to poimen-graphql for all reads and writes:
|
|
//
|
|
// - Reads: hybridSearch / entity queries
|
|
// - Writes: upsertEntity mutation (direct, no LLM extraction)
|
|
//
|
|
// 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"
|
|
"net/http"
|
|
"os"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
|
|
"forgejo.riotpiao.com/rock/homelab-frontend/internal/tracing"
|
|
)
|
|
|
|
const (
|
|
defaultGraphQLURL = "http://poimen-graphql.poimen.svc.cluster.local:8090/graphql"
|
|
)
|
|
|
|
// Handler handles session REST endpoints via internal service calls.
|
|
// All traffic (reads and writes) goes to graphqlURL (poimen-graphql).
|
|
type Handler struct {
|
|
graphqlURL string
|
|
client *http.Client
|
|
}
|
|
|
|
// NewHandler creates a Handler reading upstream URL from env var.
|
|
// GRAPHQL_UPSTREAM_URL — default: defaultGraphQLURL
|
|
func NewHandler() *Handler {
|
|
gql := os.Getenv("GRAPHQL_UPSTREAM_URL")
|
|
if gql == "" {
|
|
gql = defaultGraphQLURL
|
|
}
|
|
return &Handler{
|
|
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)
|
|
|
|
summaryBytes, err := json.Marshal(map[string]any{
|
|
"id": sessionID,
|
|
"name": req.Name,
|
|
"goal": req.Goal,
|
|
"status": "active",
|
|
"cwd": req.Cwd,
|
|
"memory_project": project,
|
|
"created_at": now,
|
|
})
|
|
if err != nil {
|
|
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "marshal failed: " + err.Error()})
|
|
return
|
|
}
|
|
if _, err := h.upsertEntity(r, project, sessionID, "SESSION", string(summaryBytes)); err != nil {
|
|
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "upsert 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)
|
|
}
|
|
|
|
// entities query (not hybridSearch) — project arg is optional, so it queries
|
|
// across ALL projects. Required because each session lives in its own
|
|
// session-<uuid> project; there is no single project to hybridSearch against.
|
|
gqlQuery := `query($q:String!,$limit:Int!){
|
|
entities(query:$q,limit:$limit,entityType:SESSION){
|
|
entities{id name summary projectId 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)
|
|
project := "session-" + sessionID
|
|
|
|
// Read current summary so the upsert carries all existing fields.
|
|
current, err := h.fetchEntitySummary(r, project, sessionID, "SESSION")
|
|
if err != nil {
|
|
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "fetch failed: " + err.Error()})
|
|
return
|
|
}
|
|
current["updated_at"] = now
|
|
if req.Name != nil {
|
|
current["name"] = *req.Name
|
|
}
|
|
if req.Goal != nil {
|
|
current["goal"] = *req.Goal
|
|
}
|
|
if req.Status != nil {
|
|
current["status"] = *req.Status
|
|
if *req.Status == "archived" {
|
|
current["archived_at"] = now
|
|
}
|
|
}
|
|
summaryBytes, err := json.Marshal(current)
|
|
if err != nil {
|
|
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "marshal failed: " + err.Error()})
|
|
return
|
|
}
|
|
|
|
if _, err := h.upsertEntity(r, project, sessionID, "SESSION", string(summaryBytes)); err != nil {
|
|
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "upsert 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
|
|
}
|
|
summaryBytes, err := json.Marshal(msContent)
|
|
if err != nil {
|
|
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "marshal failed: " + err.Error()})
|
|
return
|
|
}
|
|
if _, err := h.upsertEntity(r, project, milestoneID, "MILESTONE", string(summaryBytes)); err != nil {
|
|
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "upsert 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 ───────────────────────────────────────────────────────────────────
|
|
|
|
// upsertEntity runs a GraphQL upsertEntity mutation on poimen-graphql.
|
|
// Writes are direct (no LLM extraction). Returns the entity id assigned by the server.
|
|
// The name field is used as the stable upsert key within (project, entityType).
|
|
func (h *Handler) upsertEntity(r *http.Request, project, name, entityType, summary string) (string, error) {
|
|
mutation := `mutation($input:UpsertEntityInput!){
|
|
upsertEntity(input:$input){ id created }
|
|
}`
|
|
vars := map[string]any{
|
|
"input": map[string]any{
|
|
"project": project,
|
|
"name": name,
|
|
"entityType": entityType,
|
|
"summary": summary,
|
|
},
|
|
}
|
|
result, err := h.gql(r, mutation, vars)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
data, _ := result.(map[string]any)
|
|
d, _ := data["data"].(map[string]any)
|
|
entity, _ := d["upsertEntity"].(map[string]any)
|
|
id, _ := entity["id"].(string)
|
|
return id, nil
|
|
}
|
|
|
|
// fetchEntitySummary fetches the current summary blob of an entity by (project, name, entityType).
|
|
// Returns a parsed map so callers can merge patch fields before re-upserting.
|
|
// Returns empty map (not error) if no entity found.
|
|
func (h *Handler) fetchEntitySummary(r *http.Request, project, name, entityType string) (map[string]any, error) {
|
|
gqlQuery := `query($q:String!,$project:String!,$entityType:GqlEntityType!,$limit:Int!){
|
|
hybridSearch(query:$q,project:$project,limit:$limit,entityType:$entityType){
|
|
entities{id name summary tCreated}
|
|
}
|
|
}`
|
|
vars := map[string]any{
|
|
"q": name,
|
|
"project": project,
|
|
"entityType": entityType,
|
|
"limit": 1,
|
|
}
|
|
result, err := h.gql(r, gqlQuery, vars)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
data, _ := result.(map[string]any)
|
|
d, _ := data["data"].(map[string]any)
|
|
search, _ := d["hybridSearch"].(map[string]any)
|
|
entities, _ := search["entities"].([]any)
|
|
if len(entities) == 0 {
|
|
return map[string]any{}, nil
|
|
}
|
|
entity, _ := entities[0].(map[string]any)
|
|
summaryStr, _ := entity["summary"].(string)
|
|
var summary map[string]any
|
|
if summaryStr != "" {
|
|
if err := json.Unmarshal([]byte(summaryStr), &summary); err != nil {
|
|
return map[string]any{}, nil // corrupt summary — start fresh
|
|
}
|
|
}
|
|
if summary == nil {
|
|
summary = map[string]any{}
|
|
}
|
|
return summary, 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
|
|
}
|