diff --git a/graph.go b/graph.go new file mode 100644 index 0000000..dccb303 --- /dev/null +++ b/graph.go @@ -0,0 +1,201 @@ +package memmesh + +import ( + "context" + "net/url" + "strconv" +) + +// GraphService reads the knowledge graph built from observed memory — the +// structural half of what MemMesh stores. +// +// Observing text doesn't only produce embeddable rows; extraction also resolves +// entities and writes typed edges between them. That graph is what reaches a +// fact no single memory states outright ("who does Sarah report to?" answered +// from sarah -[member_of]-> team plus team -[led_by]-> priya). +// +// Every route here is admin-tier (/admin/memory/...); a project-scoped key gets +// a 403. +// +// Read-only by design. Entities and edges are written by extraction when you +// call Observe; the server's manual create/retire routes exist for annotation +// tooling, and exposing them here would invite hand-maintained graphs — the +// work the engine exists to do for you. +type GraphService struct{ c *Client } + +// MemoryEntity is a resolved thing — person, org, product, concept — filed +// under CanonicalName, with Aliases resolving to it. +type MemoryEntity struct { + ID string `json:"id"` + ProjectID string `json:"projectId,omitempty"` + // BrainID is the brain that FIRST created this entity. Entities dedupe per + // project, so this is provenance, not an isolation key — brain-scoped graph + // work filters on the edge's brain, which the read routes apply server-side. + BrainID string `json:"brainId,omitempty"` + Scope string `json:"scope,omitempty"` + Type string `json:"type,omitempty"` + CanonicalName string `json:"canonicalName"` + Aliases []string `json:"aliases,omitempty"` + Description string `json:"description,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` + ValidFrom string `json:"validFrom,omitempty"` + ValidTo string `json:"validTo,omitempty"` // empty while current + SupersededByID string `json:"supersededById,omitempty"` +} + +// GraphTraversalEdge is an edge as the READ routes return it — hydrated, not +// the raw memory_edge row. Subject and Object are resolved entities rather than +// ids, plus a Hop counter. +// +// This is the server's GraphTraversalEdge, returned by ListEdges, Traverse, and +// the Edges of GetEntity. The raw row shape (subjectId / objectId) is not +// exposed by any read route, so it is deliberately not modelled — a type +// nothing returns is a trap. +type GraphTraversalEdge struct { + ID string `json:"id"` + Subject MemoryEntity `json:"subject"` + // Predicate is the relationship — works_at, owns, located_in, ... + Predicate string `json:"predicate"` + // Object is nil when ObjectLiteral carries the value instead. + Object *MemoryEntity `json:"object,omitempty"` + ObjectLiteral string `json:"objectLiteral,omitempty"` + Weight float64 `json:"weight"` + ValidFrom string `json:"validFrom,omitempty"` + ValidTo string `json:"validTo,omitempty"` + SourceMemoryID string `json:"sourceMemoryId,omitempty"` + // Hop is the distance from the seed entity on a Traverse — 1 for a direct + // neighbour. ListEdges has no seed, so every edge comes back with Hop 0. + Hop int `json:"hop"` +} + +// ExtractionState reports whether KG extraction is on, platform-wide and for +// this project. +type ExtractionState struct { + PlatformEnabled bool `json:"platformEnabled"` + ProjectEnabled bool `json:"projectEnabled"` +} + +// GraphStats holds aggregate graph counts. +// +// MemoriesWithEdges against your total memory count is the useful ratio: it +// says how much of what you remember made it into the graph rather than +// remaining an isolated embedding. A low ratio usually means extraction is off +// — check Extraction before concluding the corpus simply had no relations. +type GraphStats struct { + EntityCount int64 `json:"entityCount"` + EdgeCount int64 `json:"edgeCount"` + // MemoriesWithEdges counts distinct memories that produced at least one edge. + MemoriesWithEdges int64 `json:"memoriesWithEdges"` + RetiredEntities int64 `json:"retiredEntities"` + RetiredEdges int64 `json:"retiredEdges"` + EntitiesByType map[string]int64 `json:"entitiesByType,omitempty"` + Extraction *ExtractionState `json:"extraction,omitempty"` +} + +// EntityWithEdges is an entity plus its 1-hop neighbourhood. +type EntityWithEdges struct { + Entity *MemoryEntity `json:"entity"` + Edges []GraphTraversalEdge `json:"edges"` +} + +// ListEntitiesParams filters ListEntities. Zero values are omitted. +type ListEntitiesParams struct { + Type string + Scope string + // Search is a substring match against CanonicalName and every alias. + Search string + Limit int + Offset int +} + +// TraverseParams tunes a walk. Zero values are omitted. +type TraverseParams struct { + // Hops out from the seed entity, 1-3. + Hops int + // Predicates restricts the walk, e.g. []string{"member_of", "led_by"}. + Predicates []string + AsOf string +} + +// Stats returns aggregate counts for the whole graph. +// +// Prefer this over len(ListEntities(...)) for any "how big is it" question: +// these are SQL COUNT(*)s over the full table, where the list routes page and +// would report the page size as the total. +func (s *GraphService) Stats(ctx context.Context) (*GraphStats, error) { + var out GraphStats + err := s.c.do(ctx, "GET", "/admin/memory/graph/stats", nil, nil, &out) + return &out, err +} + +// ListEntities returns entities filtered by type/scope or a substring of name +// or alias. +func (s *GraphService) ListEntities(ctx context.Context, p ListEntitiesParams) ([]MemoryEntity, error) { + q := url.Values{} + if p.Type != "" { + q.Set("type", p.Type) + } + if p.Scope != "" { + q.Set("scope", p.Scope) + } + if p.Search != "" { + q.Set("search", p.Search) + } + if p.Limit > 0 { + q.Set("limit", strconv.Itoa(p.Limit)) + } + if p.Offset > 0 { + q.Set("offset", strconv.Itoa(p.Offset)) + } + var out []MemoryEntity + err := s.c.do(ctx, "GET", "/admin/memory/entities", q, nil, &out) + return out, err +} + +// GetEntity returns one entity plus its 1-hop neighbourhood. asOf may be empty. +func (s *GraphService) GetEntity(ctx context.Context, entityID, asOf string) (*EntityWithEdges, error) { + q := url.Values{} + if asOf != "" { + q.Set("asOf", asOf) + } + var out EntityWithEdges + err := s.c.do(ctx, "GET", "/admin/memory/entities/"+url.PathEscape(entityID), q, nil, &out) + return &out, err +} + +// ListEdges returns every currently-valid edge. Use it to render a whole small +// graph; for a large one, seed from an entity and Traverse instead. Both +// arguments are optional (empty string / 0). +func (s *GraphService) ListEdges(ctx context.Context, asOf string, limit int) ([]GraphTraversalEdge, error) { + q := url.Values{} + if asOf != "" { + q.Set("asOf", asOf) + } + if limit > 0 { + q.Set("limit", strconv.Itoa(limit)) + } + var out []GraphTraversalEdge + err := s.c.do(ctx, "GET", "/admin/memory/graph/edges", q, nil, &out) + return out, err +} + +// Traverse walks out from a seed entity. +// +// This is the multi-hop path: the edges returned here connect facts no single +// memory states together, which is how a question gets answered from a chain +// rather than from one lucky vector hit. +func (s *GraphService) Traverse(ctx context.Context, entityID string, p TraverseParams) ([]GraphTraversalEdge, error) { + body := map[string]any{"entityId": entityID} + if p.Hops > 0 { + body["hops"] = p.Hops + } + if len(p.Predicates) > 0 { + body["predicates"] = p.Predicates + } + if p.AsOf != "" { + body["asOf"] = p.AsOf + } + var out []GraphTraversalEdge + err := s.c.do(ctx, "POST", "/admin/memory/graph/traverse", nil, body, &out) + return out, err +} diff --git a/graph_test.go b/graph_test.go new file mode 100644 index 0000000..e52e6ae --- /dev/null +++ b/graph_test.go @@ -0,0 +1,261 @@ +package memmesh + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +// capture records the one request a stub server saw, so tests can assert on +// route, query, and body without a live API. +type capture struct { + method string + path string + query string + body map[string]any +} + +func stub(t *testing.T, response string) (*Client, *capture) { + t.Helper() + got := &capture{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got.method = r.Method + got.path = r.URL.Path + got.query = r.URL.RawQuery + if r.Body != nil { + _ = json.NewDecoder(r.Body).Decode(&got.body) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(response)) + })) + t.Cleanup(srv.Close) + return New("sk-test", "proj_1", WithBaseURL(srv.URL)), got +} + +// ── graph ─────────────────────────────────────────────────────────────────── + +func TestGraphStatsParsesCounts(t *testing.T) { + c, got := stub(t, `{"entityCount":12142,"edgeCount":287698,"memoriesWithEdges":184737, + "retiredEntities":0,"retiredEdges":2,"entitiesByType":{"concept":5463}, + "extraction":{"platformEnabled":true,"projectEnabled":false}}`) + + st, err := c.Graph.Stats(context.Background()) + if err != nil { + t.Fatal(err) + } + if st.EntityCount != 12142 || st.EdgeCount != 287698 || st.MemoriesWithEdges != 184737 { + t.Fatalf("counts: %+v", st) + } + if st.EntitiesByType["concept"] != 5463 { + t.Fatalf("byType: %+v", st.EntitiesByType) + } + if st.Extraction == nil || st.Extraction.ProjectEnabled { + t.Fatalf("extraction: %+v", st.Extraction) + } + if got.path != "/api/v1/projects/proj_1/admin/memory/graph/stats" { + t.Fatalf("path: %s", got.path) + } +} + +func TestGraphListEntitiesSendsOnlySetFilters(t *testing.T) { + c, got := stub(t, `[]`) + if _, err := c.Graph.ListEntities(context.Background(), + ListEntitiesParams{Search: "Sarah", Limit: 5}); err != nil { + t.Fatal(err) + } + q := got.query + if !contains(q, "search=Sarah") || !contains(q, "limit=5") { + t.Fatalf("query missing filters: %s", q) + } + if contains(q, "scope=") || contains(q, "offset=") { + t.Fatalf("query carries unset filters: %s", q) + } +} + +func TestGraphListEntitiesWithoutFiltersSendsNoQuery(t *testing.T) { + c, got := stub(t, `[]`) + if _, err := c.Graph.ListEntities(context.Background(), ListEntitiesParams{}); err != nil { + t.Fatal(err) + } + if got.query != "" { + t.Fatalf("expected no query, got %q", got.query) + } +} + +func TestGraphListEntitiesEscapesFilterValues(t *testing.T) { + // An unescaped & would truncate the filter server-side and quietly return + // the wrong page — a correctness test, not a style one. + c, got := stub(t, `[]`) + if _, err := c.Graph.ListEntities(context.Background(), + ListEntitiesParams{Search: "a&b c"}); err != nil { + t.Fatal(err) + } + if !contains(got.query, "search=a%26b+c") { + t.Fatalf("value not escaped: %s", got.query) + } +} + +func TestGraphListEdgesDecodesHydratedShape(t *testing.T) { + // Regression: the read routes return GraphTraversalEdge, not the raw + // memory_edge row. There is no subjectId on the wire at all. + c, _ := stub(t, `[{"id":"g1", + "subject":{"id":"e1","canonicalName":"NVIDIA CORP","type":"org"}, + "predicate":"reported_metric", + "object":{"id":"e2","canonicalName":"Cost of Revenue"}, + "objectLiteral":null,"weight":0.85,"sourceMemoryId":"m1","hop":0}]`) + + edges, err := c.Graph.ListEdges(context.Background(), "", 1) + if err != nil { + t.Fatal(err) + } + if edges[0].Subject.CanonicalName != "NVIDIA CORP" { + t.Fatalf("subject: %+v", edges[0].Subject) + } + if edges[0].Object == nil || edges[0].Object.CanonicalName != "Cost of Revenue" { + t.Fatalf("object: %+v", edges[0].Object) + } + if edges[0].Hop != 0 || edges[0].Weight != 0.85 { + t.Fatalf("hop/weight: %+v", edges[0]) + } +} + +func TestGraphListEdgesDecodesLiteralObject(t *testing.T) { + // Object is null when the value is a literal rather than an entity. + c, _ := stub(t, `[{"id":"g2","subject":{"id":"e1","canonicalName":"NVIDIA CORP"}, + "predicate":"ticker_symbol","object":null,"objectLiteral":"NVDA","weight":0.85,"hop":0}]`) + + edges, err := c.Graph.ListEdges(context.Background(), "", 0) + if err != nil { + t.Fatal(err) + } + if edges[0].Object != nil { + t.Fatalf("expected nil object, got %+v", edges[0].Object) + } + if edges[0].ObjectLiteral != "NVDA" { + t.Fatalf("literal: %q", edges[0].ObjectLiteral) + } +} + +func TestGraphTraversePostsEntityIDAndOmitsUnset(t *testing.T) { + c, got := stub(t, `[]`) + if _, err := c.Graph.Traverse(context.Background(), "e1", + TraverseParams{Hops: 2, Predicates: []string{"member_of", "led_by"}}); err != nil { + t.Fatal(err) + } + if got.method != "POST" || got.path != "/api/v1/projects/proj_1/admin/memory/graph/traverse" { + t.Fatalf("route: %s %s", got.method, got.path) + } + if got.body["entityId"] != "e1" { + t.Fatalf("entityId: %v", got.body["entityId"]) + } + if got.body["hops"] != float64(2) { + t.Fatalf("hops: %v", got.body["hops"]) + } + if _, ok := got.body["asOf"]; ok { + t.Fatalf("asOf should be omitted: %v", got.body) + } +} + +func TestGraphGetEntityReturnsHydratedEdges(t *testing.T) { + c, _ := stub(t, `{"entity":{"id":"e1","canonicalName":"Sarah"}, + "edges":[{"id":"g1","subject":{"id":"e1","canonicalName":"Sarah"}, + "predicate":"works_at","object":{"id":"e2","canonicalName":"Acme"},"weight":0.9,"hop":1}]}`) + + hood, err := c.Graph.GetEntity(context.Background(), "e1", "") + if err != nil { + t.Fatal(err) + } + if hood.Entity == nil || hood.Entity.CanonicalName != "Sarah" { + t.Fatalf("entity: %+v", hood.Entity) + } + if hood.Edges[0].Hop != 1 { + t.Fatalf("hop: %+v", hood.Edges[0]) + } +} + +// ── observe ───────────────────────────────────────────────────────────────── + +func TestObserveTextRoutesThroughEngineAndCarriesIdentity(t *testing.T) { + c, got := stub(t, `{"saved":[{"id":"m1","type":"fact","content":"x"}],"candidateCount":3}`) + + res, err := c.Memory.Observe(context.Background(), Observe{ + Text: "I just moved to Denver.", + UserID: "user-123", + AgentID: "agent-9", + SessionID: "thread-456", + }) + if err != nil { + t.Fatal(err) + } + if got.path != "/api/v1/projects/proj_1/memory/observe" { + t.Fatalf("path: %s", got.path) + } + if got.body["text"] != "I just moved to Denver." || got.body["role"] != "user" { + t.Fatalf("body: %v", got.body) + } + if got.body["userId"] != "user-123" || got.body["agentId"] != "agent-9" || + got.body["sessionId"] != "thread-456" { + t.Fatalf("identity not forwarded: %v", got.body) + } + if len(res.Saved) != 1 || res.CandidateCount != 3 { + t.Fatalf("response: %+v", res) + } +} + +func TestObserveTextOmitsIdentityWhenUnset(t *testing.T) { + c, got := stub(t, `{"saved":[],"candidateCount":0}`) + if _, err := c.Memory.Observe(context.Background(), Observe{Text: "hello"}); err != nil { + t.Fatal(err) + } + for _, k := range []string{"userId", "agentId", "sessionId"} { + if _, ok := got.body[k]; ok { + t.Fatalf("%s should be omitted: %v", k, got.body) + } + } +} + +func TestObserveFillerReturnsEmptySavedAsSuccess(t *testing.T) { + // An empty Saved is the engine working, not an error. + c, _ := stub(t, `{"saved":[],"candidateCount":0}`) + res, err := c.Memory.Observe(context.Background(), Observe{Text: "ok thanks"}) + if err != nil { + t.Fatal(err) + } + if len(res.Saved) != 0 || res.CandidateCount != 0 { + t.Fatalf("response: %+v", res) + } +} + +func TestObserveLegacyContentStillHitsAdminMemory(t *testing.T) { + // The verbatim path must keep working, wrapped in the new response shape. + c, got := stub(t, `{"id":"m1","type":"event","content":"Ordered pizza"}`) + res, err := c.Memory.Observe(context.Background(), Observe{ + Subject: Subject{Kind: "contact", ExternalID: "sarah"}, + Content: "Ordered pizza", + }) + if err != nil { + t.Fatal(err) + } + if got.path != "/api/v1/projects/proj_1/admin/memory" { + t.Fatalf("path: %s", got.path) + } + if len(res.Saved) != 1 || res.CandidateCount != 1 { + t.Fatalf("response: %+v", res) + } +} + +func contains(haystack, needle string) bool { + return len(haystack) >= len(needle) && (haystack == needle || + len(needle) > 0 && indexOf(haystack, needle) >= 0) +} + +func indexOf(h, n string) int { + for i := 0; i+len(n) <= len(h); i++ { + if h[i:i+len(n)] == n { + return i + } + } + return -1 +} diff --git a/memmesh.go b/memmesh.go index 4eb995e..f5c7bb2 100644 --- a/memmesh.go +++ b/memmesh.go @@ -31,6 +31,7 @@ type Client struct { http *http.Client Memory *MemoryService + Graph *GraphService Lattice *LatticeService Context *ContextService Events *EventsService @@ -63,6 +64,7 @@ func New(apiKey, projectID string, opts ...Option) *Client { o(c) } c.Memory = &MemoryService{c: c} + c.Graph = &GraphService{c: c} c.Lattice = &LatticeService{c: c} c.Context = &ContextService{c: c} c.Events = &EventsService{c: c} diff --git a/memory.go b/memory.go index 3cc9399..f4dbf28 100644 --- a/memory.go +++ b/memory.go @@ -16,14 +16,36 @@ type MemoryService struct{ c *Client } // Observe records that something happened. The engine mines, wires the graph, // and revises beliefs server-side. type Observe struct { - Subject Subject `json:"-"` - Content string `json:"-"` - Type string `json:"-"` // default "event" - Scope string `json:"-"` // default "project" - Importance int `json:"-"` // default 5 - Category string `json:"-"` - ActivityType string `json:"-"` - OccurredAt string `json:"-"` + // Text is the raw message turn — the PRIMARY field. Send the whole turn + // verbatim; the engine runs extraction and keeps only what is worth + // remembering, dropping filler. When set, Observe posts to /memory/observe + // and the structured fields below are ignored — the engine resolves them + // during extraction. + Text string `json:"-"` + // Role is who said Text — defaults to "user". Only used on the Text path. + Role string `json:"-"` + // UserID is the end user this turn belongs to — your own identifier, not a + // MemMesh one. Recorded as provenance on whatever the engine keeps. + // + // NOT a tenancy boundary: search filters chatIdentityId IS NULL OR = $1, + // permissively by design, so project-wide memories stay visible to every + // caller. Isolating one end user's memories needs a project per tenant. + UserID string `json:"-"` + // AgentID is the agent or assistant that produced this turn. Provenance only. + AgentID string `json:"-"` + // SessionID is a conversation/thread id, so turns from one session stay linkable. + SessionID string `json:"-"` + + Subject Subject `json:"-"` + // Content is DEPRECATED — a pre-decided fact stored verbatim, bypassing + // extraction. Prefer Text and let the engine decide what to keep. + Content string `json:"-"` + Type string `json:"-"` // default "event" + Scope string `json:"-"` // default "project" + Importance int `json:"-"` // default 5 + Category string `json:"-"` + ActivityType string `json:"-"` + OccurredAt string `json:"-"` // Metadata carries structured fields the mining engine reads off the event. // The RFM Monetary score sums a numeric "amount" (or "value"/"total", or a // "lineItems" array) — a price written only into Content is not parsed, so @@ -70,10 +92,52 @@ func (o Observe) body() map[string]any { } // Observe ingests an event-shaped memory (the primary agent ingestion call). -func (s *MemoryService) Observe(ctx context.Context, o Observe) (*MemoryItem, error) { - var out MemoryItem - err := s.c.do(ctx, "POST", "/admin/memory", nil, o.body(), &out) - return &out, err +// Observe records a turn. +// +// PRIMARY path: set Text and the raw turn goes to the engine's Observe pipeline +// (extract -> dedupe -> graph -> embed), which returns what it chose to keep. +// Filler comes back with an empty Saved — that is success, not an error. +// CandidateCount is what extraction proposed before the dedupe/budget pass, so +// len(Saved) <= CandidateCount. +// +// LEGACY path: set Content instead and the pre-decided fact is stored verbatim, +// bypassing extraction. Wrapped in the same response shape so callers do not +// branch on which path ran. +func (s *MemoryService) Observe(ctx context.Context, o Observe) (*ObserveResponse, error) { + if strings.TrimSpace(o.Text) != "" { + var out ObserveResponse + err := s.c.do(ctx, "POST", "/memory/observe", nil, o.textBody(), &out) + return &out, err + } + var item MemoryItem + if err := s.c.do(ctx, "POST", "/admin/memory", nil, o.body(), &item); err != nil { + return nil, err + } + return &ObserveResponse{Saved: []MemoryItem{item}, CandidateCount: 1}, nil +} + +// textBody builds the /memory/observe payload. The identity fields are +// provenance; they are omitted rather than sent as null so a turn without them +// is indistinguishable from one made by an older client. +func (o Observe) textBody() map[string]any { + role := o.Role + if role == "" { + role = "user" + } + b := map[string]any{"text": o.Text, "role": role} + if o.OccurredAt != "" { + b["occurredAt"] = o.OccurredAt + } + if o.UserID != "" { + b["userId"] = o.UserID + } + if o.AgentID != "" { + b["agentId"] = o.AgentID + } + if o.SessionID != "" { + b["sessionId"] = o.SessionID + } + return b } // IngestMedia ingests an image / audio / document. The engine extracts text diff --git a/types.go b/types.go index be8961c..ad6ecda 100644 --- a/types.go +++ b/types.go @@ -125,3 +125,12 @@ type GraphEdge struct { ValidFrom string `json:"validFrom"` ValidTo *string `json:"validTo"` } + +// ObserveResponse is what Observe returns: the memories the engine chose to +// keep (empty when the turn was filler — still a success) plus how many +// candidates extraction found before the dedupe/budget pass. +// len(Saved) <= CandidateCount. +type ObserveResponse struct { + Saved []MemoryItem `json:"saved"` + CandidateCount int `json:"candidateCount"` +}