mirror of
https://github.com/hanzoai/agent.git
synced 2026-09-11 21:08:51 +00:00
A conversation is the member's who opened it
Every thread in an org listed for every member and any id could be read or appended to. A conversation now records the user who opened it and lists, opens and takes turns for them alone; one recorded before users were kept stays the org's.
This commit is contained in:
@@ -0,0 +1,49 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/hanzoai/orm"
|
||||
)
|
||||
|
||||
// A thread lists and opens for the member who opened it; a thread from before
|
||||
// users were kept stays the org's.
|
||||
func TestConversationsAreTheMembers(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
s := newStore(t.TempDir())
|
||||
defer s.closeAll()
|
||||
const org = "acme"
|
||||
alice, err := s.loadOrCreateConversation(ctx, org, "alice", "", "alice asks")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := s.loadOrCreateConversation(ctx, org, "bob", "", "bob asks"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := s.loadOrCreateConversation(ctx, org, "", "", "before users were kept"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
mine, err := s.listConversations(ctx, org, "bob")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(mine) != 2 {
|
||||
t.Fatalf("bob lists %d; want his own and the org's", len(mine))
|
||||
}
|
||||
for _, cv := range mine {
|
||||
if cv.Id() == alice.Id() {
|
||||
t.Fatalf("bob lists alice's thread %q", cv.Title)
|
||||
}
|
||||
}
|
||||
if _, err := s.conversationMessages(ctx, org, "bob", alice.Id()); !errors.Is(err, orm.ErrNotFound) {
|
||||
t.Fatalf("bob reads alice's thread: err=%v", err)
|
||||
}
|
||||
if _, err := s.loadOrCreateConversation(ctx, org, "bob", alice.Id(), "x"); !errors.Is(err, orm.ErrNotFound) {
|
||||
t.Fatalf("bob appends to alice's thread: err=%v", err)
|
||||
}
|
||||
if _, err := s.conversationMessages(ctx, org, "alice", alice.Id()); err != nil {
|
||||
t.Fatalf("alice reads her own: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"time"
|
||||
|
||||
openai "github.com/hanzoai/go-openai"
|
||||
"github.com/hanzoai/orm"
|
||||
"github.com/zap-proto/zip"
|
||||
)
|
||||
|
||||
@@ -82,9 +83,9 @@ func (s *Service) handleRun(c *zip.Ctx) error {
|
||||
|
||||
ctx := c.Context()
|
||||
|
||||
conv, err := s.store.loadOrCreateConversation(ctx, p.Org, body.ConversationID, firstUserContent(body.Messages))
|
||||
conv, err := s.store.loadOrCreateConversation(ctx, p.Org, p.User, body.ConversationID, firstUserContent(body.Messages))
|
||||
if err != nil {
|
||||
return zip.Errorf(http.StatusInternalServerError, "agent: conversation: %v", err)
|
||||
return conversationRefusal(err)
|
||||
}
|
||||
|
||||
if last := lastUserMessage(body.Messages); last != "" {
|
||||
@@ -175,7 +176,7 @@ func (s *Service) handleListConversations(c *zip.Ctx) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
items, err := s.store.listConversations(c.Context(), p.Org)
|
||||
items, err := s.store.listConversations(c.Context(), p.Org, p.User)
|
||||
if err != nil {
|
||||
return zip.Errorf(http.StatusInternalServerError, "agent: list: %v", err)
|
||||
}
|
||||
@@ -221,9 +222,9 @@ func (s *Service) handleRecord(c *zip.Ctx) error {
|
||||
}
|
||||
|
||||
ctx := c.Context()
|
||||
conv, err := s.store.loadOrCreateConversation(ctx, p.Org, body.ConversationID, firstUserContent(body.Messages))
|
||||
conv, err := s.store.loadOrCreateConversation(ctx, p.Org, p.User, body.ConversationID, firstUserContent(body.Messages))
|
||||
if err != nil {
|
||||
return zip.Errorf(http.StatusInternalServerError, "agent: conversation: %v", err)
|
||||
return conversationRefusal(err)
|
||||
}
|
||||
for _, m := range body.Messages {
|
||||
role := strings.TrimSpace(m.Role)
|
||||
@@ -237,6 +238,15 @@ func (s *Service) handleRecord(c *zip.Ctx) error {
|
||||
return c.JSON(http.StatusOK, map[string]any{"conversationId": conv.Id()})
|
||||
}
|
||||
|
||||
// conversationRefusal turns a store error into the answer a caller gets: another
|
||||
// member's thread reads as absent, anything else as the failure it is.
|
||||
func conversationRefusal(err error) error {
|
||||
if errors.Is(err, orm.ErrNotFound) {
|
||||
return zip.ErrNotFound("conversation not found")
|
||||
}
|
||||
return zip.Errorf(http.StatusInternalServerError, "agent: conversation: %v", err)
|
||||
}
|
||||
|
||||
type msgOut struct {
|
||||
ID string `json:"id"`
|
||||
Role string `json:"role"`
|
||||
@@ -254,9 +264,9 @@ func (s *Service) handleConversation(c *zip.Ctx) error {
|
||||
if id == "" {
|
||||
return zip.ErrBadRequest("conversation id required")
|
||||
}
|
||||
msgs, err := s.store.conversationMessages(c.Context(), p.Org, id)
|
||||
msgs, err := s.store.conversationMessages(c.Context(), p.Org, p.User, id)
|
||||
if err != nil {
|
||||
return zip.Errorf(http.StatusInternalServerError, "agent: messages: %v", err)
|
||||
return conversationRefusal(err)
|
||||
}
|
||||
out := make([]msgOut, 0, len(msgs))
|
||||
for _, m := range msgs {
|
||||
|
||||
@@ -17,10 +17,13 @@ import (
|
||||
|
||||
// Conversation is one persisted chat thread. Org is the owning org — physical
|
||||
// isolation (one SQLite file per org) already scopes it; Org is stored for clarity
|
||||
// and defense-in-depth.
|
||||
// and defense-in-depth. User is the member who opened it: a thread is one
|
||||
// person's inside a shared org, so it lists and opens for them alone. A thread
|
||||
// recorded before users were kept has none, and stays the org's.
|
||||
type Conversation struct {
|
||||
orm.Model[Conversation]
|
||||
Org string `json:"org"`
|
||||
User string `json:"user"`
|
||||
Title string `json:"title"`
|
||||
}
|
||||
|
||||
@@ -130,7 +133,7 @@ func orgSlug(org string) (string, error) {
|
||||
// loadOrCreateConversation returns the conversation for id (in the org's DB),
|
||||
// creating a fresh one when id is empty or not found. Physical per-org isolation
|
||||
// means a found row always belongs to org; the Org check is belt-and-suspenders.
|
||||
func (s *store) loadOrCreateConversation(ctx context.Context, org, id, title string) (*Conversation, error) {
|
||||
func (s *store) loadOrCreateConversation(ctx context.Context, org, user, id, title string) (*Conversation, error) {
|
||||
db, err := s.dbFor(org)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -138,6 +141,9 @@ func (s *store) loadOrCreateConversation(ctx context.Context, org, id, title str
|
||||
if id = strings.TrimSpace(id); id != "" {
|
||||
conv, gerr := orm.Get[Conversation](db, id)
|
||||
if gerr == nil && conv.Org == org {
|
||||
if !owns(conv, user) {
|
||||
return nil, orm.ErrNotFound
|
||||
}
|
||||
return conv, nil
|
||||
}
|
||||
if gerr != nil && gerr != orm.ErrNotFound {
|
||||
@@ -149,6 +155,7 @@ func (s *store) loadOrCreateConversation(ctx context.Context, org, id, title str
|
||||
conv := orm.New[Conversation](db)
|
||||
conv.SetId(newID())
|
||||
conv.Org = org
|
||||
conv.User = strings.TrimSpace(user)
|
||||
conv.Title = clampTitle(title)
|
||||
if err := conv.CreateCtx(ctx); err != nil {
|
||||
return nil, err
|
||||
@@ -176,25 +183,38 @@ func (s *store) appendMessage(ctx context.Context, org, convID, role, content st
|
||||
}
|
||||
|
||||
// listConversations returns the org's conversations, most-recently-updated first.
|
||||
func (s *store) listConversations(ctx context.Context, org string) ([]*Conversation, error) {
|
||||
func (s *store) listConversations(ctx context.Context, org, user string) ([]*Conversation, error) {
|
||||
db, err := s.dbFor(org)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items, err := orm.TypedQuery[Conversation](db).GetAll(ctx)
|
||||
all, err := orm.TypedQuery[Conversation](db).GetAll(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items := all[:0]
|
||||
for _, cv := range all {
|
||||
if owns(cv, user) {
|
||||
items = append(items, cv)
|
||||
}
|
||||
}
|
||||
sort.SliceStable(items, func(i, j int) bool { return items[i].UpdatedAt.After(items[j].UpdatedAt) })
|
||||
return items, nil
|
||||
}
|
||||
|
||||
// conversationMessages returns a conversation's messages in chronological order.
|
||||
func (s *store) conversationMessages(ctx context.Context, org, convID string) ([]*Message, error) {
|
||||
func (s *store) conversationMessages(ctx context.Context, org, user, convID string) ([]*Message, error) {
|
||||
db, err := s.dbFor(org)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
conv, err := orm.Get[Conversation](db, convID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if conv.Org != org || !owns(conv, user) {
|
||||
return nil, orm.ErrNotFound
|
||||
}
|
||||
items, err := orm.TypedQuery[Message](db).Filter("ConversationId=", convID).GetAll(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -203,6 +223,12 @@ func (s *store) conversationMessages(ctx context.Context, org, convID string) ([
|
||||
return items, nil
|
||||
}
|
||||
|
||||
// owns reports whether a conversation is this member's to list and read: theirs,
|
||||
// or one recorded before users were kept.
|
||||
func owns(cv *Conversation, user string) bool {
|
||||
return cv.User == "" || cv.User == strings.TrimSpace(user)
|
||||
}
|
||||
|
||||
// clampTitle derives a short, single-line conversation title.
|
||||
func clampTitle(s string) string {
|
||||
s = strings.TrimSpace(strings.ReplaceAll(s, "\n", " "))
|
||||
|
||||
Reference in New Issue
Block a user