pathwayd: HTTP service over internal/pathway · 11/11 smoke gate
Network-callable Mem0-style trace memory at :3217, fronted by gateway
/v1/pathway/*. Closes the ADR-004 wire-up: store substrate landed in
2a6234f, this lands the HTTP surface + [pathwayd] config + acceptance
gate.
Smoke proves the architecturally distinctive properties: Revise →
History walks the predecessor chain backward (audit trail), Retire
excludes from Search default but stays Get-able, AddIdempotent bumps
replay_count without replacing — and all survive kill+restart via
JSONL log replay.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
@@ -0,0 +1,278 @@
|
||||
// pathwayd is the pathway memory service. Wraps internal/pathway's
|
||||
// Store with HTTP routes for the Mem0-style operations defined in
|
||||
// ADR-004.
|
||||
//
|
||||
// Routes (all under /pathway):
|
||||
// POST /pathway/add — new trace with fresh UID
|
||||
// POST /pathway/add_idempotent — UID-keyed add or replay-bump
|
||||
// POST /pathway/update — replace content for an existing UID
|
||||
// POST /pathway/revise — new revision linked to predecessor
|
||||
// POST /pathway/retire — mark trace retired (excluded from search)
|
||||
// GET /pathway/get/{uid} — fetch one trace (incl. retired)
|
||||
// GET /pathway/history/{uid} — backward chain via predecessor links
|
||||
// POST /pathway/search — filter-based listing
|
||||
// GET /pathway/stats — total/active/retired counters
|
||||
//
|
||||
// Persistence: optional. Empty [pathwayd].persist_path = in-memory
|
||||
// only (matches vectord G1's pattern). Set a path for durable
|
||||
// per-trace JSONL append.
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
|
||||
"git.agentview.dev/profit/golangLAKEHOUSE/internal/pathway"
|
||||
"git.agentview.dev/profit/golangLAKEHOUSE/internal/shared"
|
||||
)
|
||||
|
||||
const maxRequestBytes = 4 << 20 // 4 MiB cap on request bodies
|
||||
|
||||
func main() {
|
||||
configPath := flag.String("config", "lakehouse.toml", "path to TOML config")
|
||||
flag.Parse()
|
||||
|
||||
cfg, err := shared.LoadConfig(*configPath)
|
||||
if err != nil {
|
||||
slog.Error("config", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Persistence is optional — empty path = in-memory ephemeral.
|
||||
var persistor *pathway.Persistor
|
||||
if cfg.Pathwayd.PersistPath != "" {
|
||||
persistor, err = pathway.NewPersistor(cfg.Pathwayd.PersistPath)
|
||||
if err != nil {
|
||||
slog.Error("pathway persistor", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
store := pathway.NewStore(persistor)
|
||||
if persistor != nil {
|
||||
n, err := store.Load()
|
||||
if err != nil {
|
||||
slog.Warn("pathway load", "err", err, "loaded", n)
|
||||
} else {
|
||||
slog.Info("pathway loaded", "events", n, "path", cfg.Pathwayd.PersistPath)
|
||||
}
|
||||
}
|
||||
|
||||
h := &handlers{store: store}
|
||||
|
||||
if err := shared.Run("pathwayd", cfg.Pathwayd.Bind, h.register, cfg.Auth); err != nil {
|
||||
slog.Error("server", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
type handlers struct {
|
||||
store *pathway.Store
|
||||
}
|
||||
|
||||
func (h *handlers) register(r chi.Router) {
|
||||
r.Post("/pathway/add", h.handleAdd)
|
||||
r.Post("/pathway/add_idempotent", h.handleAddIdempotent)
|
||||
r.Post("/pathway/update", h.handleUpdate)
|
||||
r.Post("/pathway/revise", h.handleRevise)
|
||||
r.Post("/pathway/retire", h.handleRetire)
|
||||
r.Get("/pathway/get/{uid}", h.handleGet)
|
||||
r.Get("/pathway/history/{uid}", h.handleHistory)
|
||||
r.Post("/pathway/search", h.handleSearch)
|
||||
r.Get("/pathway/stats", h.handleStats)
|
||||
}
|
||||
|
||||
// ── request shapes ───────────────────────────────────────────────
|
||||
|
||||
type addRequest struct {
|
||||
Content json.RawMessage `json:"content"`
|
||||
Tags []string `json:"tags,omitempty"`
|
||||
}
|
||||
|
||||
type addIdempotentRequest struct {
|
||||
UID string `json:"uid"`
|
||||
Content json.RawMessage `json:"content"`
|
||||
Tags []string `json:"tags,omitempty"`
|
||||
}
|
||||
|
||||
type updateRequest struct {
|
||||
UID string `json:"uid"`
|
||||
Content json.RawMessage `json:"content"`
|
||||
}
|
||||
|
||||
type reviseRequest struct {
|
||||
PredecessorUID string `json:"predecessor_uid"`
|
||||
Content json.RawMessage `json:"content"`
|
||||
Tags []string `json:"tags,omitempty"`
|
||||
}
|
||||
|
||||
type retireRequest struct {
|
||||
UID string `json:"uid"`
|
||||
}
|
||||
|
||||
type searchRequest struct {
|
||||
Tag string `json:"tag,omitempty"`
|
||||
ContentContains string `json:"content_contains,omitempty"`
|
||||
CreatedAfterNs int64 `json:"created_after_ns,omitempty"`
|
||||
CreatedBeforeNs int64 `json:"created_before_ns,omitempty"`
|
||||
IncludeRetired bool `json:"include_retired,omitempty"`
|
||||
}
|
||||
|
||||
// ── handlers ────────────────────────────────────────────────────
|
||||
|
||||
func (h *handlers) handleAdd(w http.ResponseWriter, r *http.Request) {
|
||||
var req addRequest
|
||||
if !decodeJSON(w, r, &req) {
|
||||
return
|
||||
}
|
||||
tr, err := h.store.Add(req.Content, req.Tags...)
|
||||
if writeStoreError(w, err) {
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusCreated, tr)
|
||||
}
|
||||
|
||||
func (h *handlers) handleAddIdempotent(w http.ResponseWriter, r *http.Request) {
|
||||
var req addIdempotentRequest
|
||||
if !decodeJSON(w, r, &req) {
|
||||
return
|
||||
}
|
||||
tr, err := h.store.AddIdempotent(req.UID, req.Content, req.Tags...)
|
||||
if writeStoreError(w, err) {
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, tr)
|
||||
}
|
||||
|
||||
func (h *handlers) handleUpdate(w http.ResponseWriter, r *http.Request) {
|
||||
var req updateRequest
|
||||
if !decodeJSON(w, r, &req) {
|
||||
return
|
||||
}
|
||||
if err := h.store.Update(req.UID, req.Content); writeStoreError(w, err) {
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{"status": "updated"})
|
||||
}
|
||||
|
||||
func (h *handlers) handleRevise(w http.ResponseWriter, r *http.Request) {
|
||||
var req reviseRequest
|
||||
if !decodeJSON(w, r, &req) {
|
||||
return
|
||||
}
|
||||
tr, err := h.store.Revise(req.PredecessorUID, req.Content, req.Tags...)
|
||||
if writeStoreError(w, err) {
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusCreated, tr)
|
||||
}
|
||||
|
||||
func (h *handlers) handleRetire(w http.ResponseWriter, r *http.Request) {
|
||||
var req retireRequest
|
||||
if !decodeJSON(w, r, &req) {
|
||||
return
|
||||
}
|
||||
if err := h.store.Retire(req.UID); writeStoreError(w, err) {
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
func (h *handlers) handleGet(w http.ResponseWriter, r *http.Request) {
|
||||
uid := chi.URLParam(r, "uid")
|
||||
tr, err := h.store.Get(uid)
|
||||
if writeStoreError(w, err) {
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, tr)
|
||||
}
|
||||
|
||||
func (h *handlers) handleHistory(w http.ResponseWriter, r *http.Request) {
|
||||
uid := chi.URLParam(r, "uid")
|
||||
chain, err := h.store.History(uid)
|
||||
if writeStoreError(w, err) {
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"chain": chain,
|
||||
"length": len(chain),
|
||||
})
|
||||
}
|
||||
|
||||
func (h *handlers) handleSearch(w http.ResponseWriter, r *http.Request) {
|
||||
var req searchRequest
|
||||
if !decodeJSON(w, r, &req) {
|
||||
return
|
||||
}
|
||||
results := h.store.Search(pathway.SearchFilter{
|
||||
Tag: req.Tag,
|
||||
ContentContains: req.ContentContains,
|
||||
CreatedAfterNs: req.CreatedAfterNs,
|
||||
CreatedBeforeNs: req.CreatedBeforeNs,
|
||||
IncludeRetired: req.IncludeRetired,
|
||||
})
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"results": results,
|
||||
"count": len(results),
|
||||
})
|
||||
}
|
||||
|
||||
func (h *handlers) handleStats(w http.ResponseWriter, _ *http.Request) {
|
||||
writeJSON(w, http.StatusOK, h.store.Stats())
|
||||
}
|
||||
|
||||
// ── helpers ────────────────────────────────────────────────────
|
||||
|
||||
func decodeJSON(w http.ResponseWriter, r *http.Request, v any) bool {
|
||||
defer r.Body.Close()
|
||||
r.Body = http.MaxBytesReader(w, r.Body, maxRequestBytes)
|
||||
if err := json.NewDecoder(r.Body).Decode(v); err != nil {
|
||||
var maxErr *http.MaxBytesError
|
||||
if errors.As(err, &maxErr) || strings.Contains(err.Error(), "http: request body too large") {
|
||||
http.Error(w, "body too large", http.StatusRequestEntityTooLarge)
|
||||
return false
|
||||
}
|
||||
http.Error(w, "decode body: "+err.Error(), http.StatusBadRequest)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, code int, v any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(code)
|
||||
if err := json.NewEncoder(w).Encode(v); err != nil {
|
||||
slog.Warn("pathway write json", "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
// writeStoreError maps internal/pathway sentinel errors to HTTP
|
||||
// status codes. Returns true if a response was written (caller
|
||||
// should return). Returns false on success (caller continues).
|
||||
func writeStoreError(w http.ResponseWriter, err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
switch {
|
||||
case errors.Is(err, pathway.ErrNotFound):
|
||||
http.Error(w, err.Error(), http.StatusNotFound)
|
||||
case errors.Is(err, pathway.ErrPredecessorMissing):
|
||||
http.Error(w, err.Error(), http.StatusNotFound)
|
||||
case errors.Is(err, pathway.ErrEmptyUID),
|
||||
errors.Is(err, pathway.ErrInvalidContent):
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
case errors.Is(err, pathway.ErrCycle):
|
||||
http.Error(w, err.Error(), http.StatusConflict)
|
||||
default:
|
||||
slog.Error("pathway store", "err", err)
|
||||
http.Error(w, "internal", http.StatusInternalServerError)
|
||||
}
|
||||
return true
|
||||
}
|
||||
Reference in New Issue
Block a user