diff --git a/cmd/gateway_subagent_announce_queue_test.go b/cmd/gateway_subagent_announce_queue_test.go index 7d24eb44..48e21e1c 100644 --- a/cmd/gateway_subagent_announce_queue_test.go +++ b/cmd/gateway_subagent_announce_queue_test.go @@ -25,6 +25,9 @@ func (*recordingGatewayTaskStore) Get(context.Context, uuid.UUID, uuid.UUID) (*s func (*recordingGatewayTaskStore) UpdateStatus(context.Context, uuid.UUID, uuid.UUID, string, *string, int, int64, int64) error { return nil } +func (*recordingGatewayTaskStore) ListDelegationsByChat(context.Context, uuid.UUID, string) ([]store.SubagentTaskData, error) { + return nil, nil +} func (*recordingGatewayTaskStore) ListByParent(context.Context, uuid.UUID, string) ([]store.SubagentTaskData, error) { return nil, nil } diff --git a/internal/store/pg/subagent_tasks.go b/internal/store/pg/subagent_tasks.go index 97b84856..4441a437 100644 --- a/internal/store/pg/subagent_tasks.go +++ b/internal/store/pg/subagent_tasks.go @@ -204,6 +204,37 @@ func (s *PGSubagentTaskStore) ListBySession( return collectTasks(rows) } +// ListDelegationsByChat returns delegations raised in one chat (tenant-scoped). +// The inverse of the predicate ListByParent and ListBySession carry: those serve +// spawn and filter delegations out, so without this there is no way to list a +// delegation at all — only Get by ID reaches one. +func (s *PGSubagentTaskStore) ListDelegationsByChat( + ctx context.Context, rootAgentID uuid.UUID, chatID string, +) ([]store.SubagentTaskData, error) { + tid, err := requireTenantID(ctx) + if err != nil { + return nil, err + } + if rootAgentID == uuid.Nil { + return nil, store.ErrSubagentRootAgentIDRequired + } + if chatID == "" { + return nil, nil + } + + q := fmt.Sprintf(`SELECT %s FROM subagent_tasks + WHERE tenant_id = $1 AND root_agent_id = $2 AND origin_chat_id = $3 + AND COALESCE(metadata->>'completion_kind', 'subagent') = 'delegate' + ORDER BY created_at DESC LIMIT 50`, subagentTaskSelectCols) + rows, err := s.db.QueryContext(ctx, q, tid, rootAgentID, chatID) + if err != nil { + return nil, err + } + defer rows.Close() + + return collectTasks(rows) +} + // Archive marks a bounded batch of old terminal tasks as archived. func (s *PGSubagentTaskStore) Archive( ctx context.Context, rootAgentID uuid.UUID, olderThan time.Duration, limit int, diff --git a/internal/store/sqlitestore/subagent-tasks.go b/internal/store/sqlitestore/subagent-tasks.go index 08617780..95549a16 100644 --- a/internal/store/sqlitestore/subagent-tasks.go +++ b/internal/store/sqlitestore/subagent-tasks.go @@ -221,6 +221,36 @@ func (s *SQLiteSubagentTaskStore) ListBySession( return collectTasks(rows) } +// ListDelegationsByChat returns delegations raised in one chat (tenant-scoped). +// The inverse of the predicate ListByParent and ListBySession carry: those serve +// spawn and filter delegations out, so without this there is no way to list a +// delegation at all — only Get by ID reaches one. +func (s *SQLiteSubagentTaskStore) ListDelegationsByChat( + ctx context.Context, rootAgentID uuid.UUID, chatID string, +) ([]store.SubagentTaskData, error) { + tid, err := requireTenantID(ctx) + if err != nil { + return nil, err + } + if rootAgentID == uuid.Nil { + return nil, store.ErrSubagentRootAgentIDRequired + } + if chatID == "" { + return nil, nil + } + + q := fmt.Sprintf(`SELECT %s FROM subagent_tasks + WHERE tenant_id = ? AND root_agent_id = ? AND origin_chat_id = ? + AND COALESCE(json_extract(metadata, '$.completion_kind'), 'subagent') = 'delegate' + ORDER BY created_at DESC LIMIT 50`, subagentTaskSelectCols) + rows, err := s.db.QueryContext(ctx, q, tid, rootAgentID, chatID) + if err != nil { + return nil, err + } + defer rows.Close() + return collectTasks(rows) +} + // Archive marks a bounded batch of old terminal tasks as archived. func (s *SQLiteSubagentTaskStore) Archive( ctx context.Context, rootAgentID uuid.UUID, olderThan time.Duration, limit int, diff --git a/internal/store/sqlitestore/subagent-tasks_test.go b/internal/store/sqlitestore/subagent-tasks_test.go index cec9a193..eb297410 100644 --- a/internal/store/sqlitestore/subagent-tasks_test.go +++ b/internal/store/sqlitestore/subagent-tasks_test.go @@ -584,3 +584,111 @@ func assertSQLiteArchivedCount( t.Fatalf("archived tasks for %s/%s = %d, want %d", tenantID, rootAgentID, got, want) } } + +// createSQLiteDelegation persists a delegation row: completion_kind=delegate and +// an origin chat, which is what ListDelegationsByChat selects on. +func createSQLiteDelegation( + t *testing.T, + taskStore *SQLiteSubagentTaskStore, + ctx context.Context, + rootAgentID uuid.UUID, + rootAgentKey, chatID string, +) uuid.UUID { + t.Helper() + + id := uuid.Must(uuid.NewV7()) + chat := chatID + session := "session-" + chatID + task := &store.SubagentTaskData{ + RootAgentID: rootAgentID, + ParentAgentKey: rootAgentKey, + SessionKey: &session, + OriginChatID: &chat, + Subject: "Delegate to brain", + Description: "verify delegation listing", + Status: "completed", + Depth: 1, + Metadata: map[string]any{"completion_kind": "delegate"}, + } + task.ID = id + if err := taskStore.Create(ctx, task); err != nil { + t.Fatalf("Create(%s): %v", id, err) + } + return id +} + +// ListDelegationsByChat is the only listing that returns delegations at all: +// ListByParent and ListBySession both carry "completion_kind <> 'delegate'" to +// serve spawn. This pins that complementarity against the real SQL — a unit test +// over a fake store cannot see it, and the first version of the delegate list +// action shipped green and returned nothing in production because of exactly +// that gap. +func TestSQLiteListDelegationsByChatIsTheOnlyListingThatSeesDelegations(t *testing.T) { + db := newHookTestDB(t) + tenantA, rootAID := seedHookTenantAgent(t, db) + tenantB, tenantBRootID := seedHookTenantAgent(t, db) + ctxA := sqliteTenantCtx(tenantA) + ctxB := sqliteTenantCtx(tenantB) + taskStore := NewSQLiteSubagentTaskStore(db) + + const rootA = "root-a" + wanted := createSQLiteDelegation(t, taskStore, ctxA, rootAID, rootA, "chat-a") + _ = createSQLiteDelegation(t, taskStore, ctxA, rootAID, rootA, "chat-b") + spawned := createSQLiteSubagentTask(t, taskStore, ctxA, rootAID, rootA, "session-chat-a", "completed") + _ = createSQLiteDelegation(t, taskStore, ctxB, tenantBRootID, rootA, "chat-a") + + got, err := taskStore.ListDelegationsByChat(ctxA, rootAID, "chat-a") + if err != nil { + t.Fatalf("ListDelegationsByChat: %v", err) + } + if len(got) != 1 || got[0].ID != wanted { + t.Fatalf("returned %d rows (%v), want exactly the delegation from chat-a (%s)", len(got), got, wanted) + } + + // The spawn in the same chat is not a delegation. + for _, task := range got { + if task.ID == spawned { + t.Errorf("a spawned subagent leaked into the delegation listing") + } + } + + // Another tenant's delegation in a chat of the same name stays invisible. + crossTenant, err := taskStore.ListDelegationsByChat(ctxB, rootAID, "chat-a") + if err != nil { + t.Fatalf("ListDelegationsByChat(other tenant): %v", err) + } + if len(crossTenant) != 0 { + t.Errorf("listing crossed a tenant boundary: %v", crossTenant) + } + + // And the complement: the spawn-facing listings must not return it, which is + // why this method has to exist in the first place. + byParent, err := taskStore.ListByParent(ctxA, rootAID, "") + if err != nil { + t.Fatalf("ListByParent: %v", err) + } + for _, task := range byParent { + if task.ID == wanted { + t.Fatalf("ListByParent returned a delegation; if that is now intended, " + + "ListDelegationsByChat and this test need revisiting") + } + } + bySession, err := taskStore.ListBySession(ctxA, rootAID, "session-chat-a") + if err != nil { + t.Fatalf("ListBySession: %v", err) + } + for _, task := range bySession { + if task.ID == wanted { + t.Fatalf("ListBySession returned a delegation") + } + } + + // No chat, nothing listed — never a fallback to every delegation. + empty, err := taskStore.ListDelegationsByChat(ctxA, rootAID, "") + if err != nil { + t.Fatalf("ListDelegationsByChat(no chat): %v", err) + } + if len(empty) != 0 { + t.Errorf("empty chat returned %d rows", len(empty)) + } +} diff --git a/internal/store/subagent_store.go b/internal/store/subagent_store.go index 09803f69..8d9221c8 100644 --- a/internal/store/subagent_store.go +++ b/internal/store/subagent_store.go @@ -76,6 +76,12 @@ type SubagentTaskStore interface { // ListBySession returns tasks for a session owned by the tenant and immutable root-agent UUID. ListBySession(ctx context.Context, rootAgentID uuid.UUID, sessionKey string) ([]SubagentTaskData, error) + // ListDelegationsByChat returns delegation tasks raised in one chat, owned by + // the tenant and immutable root-agent UUID. ListByParent and ListBySession + // both exclude delegations on purpose — they serve spawn — so this is the + // only listing that returns them. Ordered by created_at DESC. + ListDelegationsByChat(ctx context.Context, rootAgentID uuid.UUID, chatID string) ([]SubagentTaskData, error) + // Archive marks at most limit old terminal tasks owned by the tenant and // immutable root-agent UUID as archived. Returns the number of rows affected. Archive(ctx context.Context, rootAgentID uuid.UUID, olderThan time.Duration, limit int) (int64, error) diff --git a/internal/tools/delegate_completion_ledger.go b/internal/tools/delegate_completion_ledger.go index 3f45b730..3ff52e11 100644 --- a/internal/tools/delegate_completion_ledger.go +++ b/internal/tools/delegate_completion_ledger.go @@ -236,3 +236,92 @@ func (t *DelegateTool) executeGetCompletion(ctx context.Context, args map[string } return NewResult(string(encoded)) } + +// delegateListLimit caps how many delegations action=list reports. The point is +// to re-acquire a handle you just lost, not to page through history. +const delegateListLimit = 20 + +// executeListCompletions reports the delegations started in this chat, newest +// first, so a caller that lost a delegation ID can recover it. +// +// Without this, a delegation result was addressable only by a UUID the calling +// model had to transcribe by hand across turns; one wrong character orphaned a +// completed, durably stored result with no way back (#1545). +// +// Scope is tenant (via dbCtx) and calling agent, as action=get already resolves, +// plus the origin chat. Three things follow from choosing the chat: +// +// - It survives a session reset. Deferring long work, clearing the context and +// coming back to ask for status is ordinary use; scoping by SessionKey would +// return nothing exactly then, which is when the handle is most likely gone. +// - It keeps chats apart, which is the enumeration boundary #1525 is about: +// there, spawn's list filters on the parent agent key alone and ignores the +// session, so one chat reads another chat's task text. +// - It does not carry a conversation between chats. A delegation started in a +// team chat stays visible in that team chat, not in someone's DM with the +// same agent. The chat is the unit of visibility, and history stays where it +// began. +// +// In a direct chat this separates users too, since the chat ID is then per +// person. Group chats deliberately show the whole group what the group started. +// +// ToolChatIDFromCtx is the same accessor createDelegateCompletion writes from, +// not OriginChatIDFromCtx — the latter prefers the workspace chat ID and would +// disagree with the stored value on delegated runs. +func (t *DelegateTool) executeListCompletions(ctx context.Context) *Result { + tenantID := store.TenantIDFromContext(ctx) + fromAgentID := store.AgentIDFromContext(ctx) + if tenantID == uuid.Nil || fromAgentID == uuid.Nil { + return ErrorResult("delegate list requires tenant and agent context") + } + if t.taskStore == nil { + return ErrorResult("durable delegation task tracking is unavailable") + } + chatID := ToolChatIDFromCtx(ctx) + if chatID == "" { + // No chat means nothing can be scoped to one. Listing every delegation + // this agent ever made would cross the boundary this predicate draws. + return ErrorResult("delegate list requires a chat context") + } + + dbCtx := store.WithTenantID(context.WithoutCancel(ctx), tenantID) + // ListDelegationsByChat, not ListByParent: the latter serves spawn and carries + // "completion_kind <> 'delegate'" in its SQL, so it can never return a + // delegation. Both the chat scope and the delegate-only predicate live in the + // query — do not re-filter here, or a change to the query goes unnoticed the + // way that one did. + tasks, err := t.taskStore.ListDelegationsByChat(dbCtx, fromAgentID, chatID) + if err != nil { + slog.Warn("delegate.list.failed", "agent_id", fromAgentID, "error", err) + return ErrorResult("failed to list delegations") + } + + items := make([]map[string]any, 0, delegateListLimit) + for i := range tasks { + task := &tasks[i] + item := map[string]any{ + "delegation_id": task.ID.String(), + "status": task.Status, + "created_at": task.CreatedAt.UTC().Format(time.RFC3339), + } + if target, ok := task.Metadata[delegateCompletionTargetKey].(string); ok && target != "" { + item["agent"] = target + } + if task.CompletedAt != nil { + item["completed_at"] = task.CompletedAt.UTC().Format(time.RFC3339) + } + items = append(items, item) + if len(items) == delegateListLimit { + break + } + } + + encoded, err := json.Marshal(map[string]any{ + "delegations": items, + "count": len(items), + }) + if err != nil { + return ErrorResult("failed to encode delegation list") + } + return NewResult(string(encoded)) +} diff --git a/internal/tools/delegate_list_test.go b/internal/tools/delegate_list_test.go new file mode 100644 index 00000000..ef9a5675 --- /dev/null +++ b/internal/tools/delegate_list_test.go @@ -0,0 +1,176 @@ +package tools + +import ( + "context" + "encoding/json" + "fmt" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// chatScopedTaskStore serves ListDelegationsByChat and records what it was asked +// for. The embedded interface is nil on purpose, and ListByParent/ListBySession +// panic outright: both carry "completion_kind <> 'delegate'" in their real SQL +// and can never return a delegation, so a regression to either must fail loudly +// here rather than quietly return nothing — which is exactly how the first +// version of this shipped green and did nothing in production. +type chatScopedTaskStore struct { + store.SubagentTaskStore + rows []store.SubagentTaskData + listCalls int + askedAgent uuid.UUID + askedChat string +} + +func (s *chatScopedTaskStore) ListDelegationsByChat( + _ context.Context, rootAgentID uuid.UUID, chatID string, +) ([]store.SubagentTaskData, error) { + s.listCalls++ + s.askedAgent = rootAgentID + s.askedChat = chatID + // Stand in for the query's own chat predicate, so the fixtures can carry + // rows from other chats and this still behaves like the real store. + var out []store.SubagentTaskData + for _, r := range s.rows { + if r.OriginChatID != nil && *r.OriginChatID == chatID && + completionKind(r.Metadata) == asyncCompletionKindDelegate { + out = append(out, r) + } + } + return out, nil +} + +func (s *chatScopedTaskStore) ListByParent(context.Context, uuid.UUID, string) ([]store.SubagentTaskData, error) { + panic("delegate list must not use ListByParent: its SQL excludes delegations") +} + +func (s *chatScopedTaskStore) ListBySession(context.Context, uuid.UUID, string) ([]store.SubagentTaskData, error) { + panic("delegate list must not use ListBySession: its SQL excludes delegations") +} + +func delegationRow(chatID, sessionKey, target string, created time.Time) store.SubagentTaskData { + return store.SubagentTaskData{ + BaseModel: store.BaseModel{ID: uuid.New(), CreatedAt: created}, + Status: TaskStatusCompleted, + SessionKey: &sessionKey, + OriginChatID: &chatID, + Metadata: map[string]any{ + asyncCompletionKindKey: asyncCompletionKindDelegate, + delegateCompletionTargetKey: target, + }, + } +} + +func spawnRow(chatID string) store.SubagentTaskData { + return store.SubagentTaskData{ + BaseModel: store.BaseModel{ID: uuid.New(), CreatedAt: time.Unix(1, 0)}, + Status: TaskStatusCompleted, + OriginChatID: &chatID, + Metadata: map[string]any{asyncCompletionKindKey: asyncCompletionKindSubagent}, + } +} + +func listCtx(chatID string) context.Context { + ctx := store.WithTenantID(context.Background(), uuid.New()) + ctx = store.WithAgentID(ctx, uuid.New()) + return WithToolChatID(ctx, chatID) +} + +func decodeList(t *testing.T, res *Result) []map[string]any { + t.Helper() + if res == nil || res.IsError { + t.Fatalf("list result = %#v", res) + } + var payload struct { + Delegations []map[string]any `json:"delegations"` + Count int `json:"count"` + } + if err := json.Unmarshal([]byte(res.ForLLM), &payload); err != nil { + t.Fatalf("decode %q: %v", res.ForLLM, err) + } + if payload.Count != len(payload.Delegations) { + t.Errorf("count %d disagrees with %d listed", payload.Count, len(payload.Delegations)) + } + return payload.Delegations +} + +func newListTool(rows []store.SubagentTaskData) (*DelegateTool, *chatScopedTaskStore) { + s := &chatScopedTaskStore{rows: rows} + tool := NewDelegateTool(nil, noopAgentCRUD{}, nil, nil) + tool.SetTaskStore(s) + return tool, s +} + +// The chat is the unit of visibility: a delegation started in one chat is not +// carried into another chat with the same agent, and spawned subagents sharing +// the table are not delegations. +func TestDelegateListIsScopedToItsChat(t *testing.T) { + tool, s := newListTool([]store.SubagentTaskData{ + delegationRow("chat-a", "session-1", "brain", time.Unix(200, 0)), + delegationRow("chat-b", "session-1", "brain", time.Unix(300, 0)), + spawnRow("chat-a"), + }) + got := decodeList(t, tool.executeListCompletions(listCtx("chat-a"))) + if s.askedChat != "chat-a" { + t.Errorf("store was asked for chat %q, want the caller's chat", s.askedChat) + } + if len(got) != 1 { + t.Fatalf("listed %d entries, want only this chat's delegation: %v", len(got), got) + } + if got[0]["agent"] != "brain" || got[0]["status"] != TaskStatusCompleted { + t.Errorf("entry = %v", got[0]) + } + if got[0]["delegation_id"] == "" || got[0]["created_at"] == "" { + t.Errorf("entry is missing the fields needed to re-acquire a handle: %v", got[0]) + } +} + +// The case the whole action exists for. Deferring long work, clearing the +// session and coming back to ask for status is ordinary use; scoping by session +// key would return nothing exactly then. +func TestDelegateListSurvivesSessionReset(t *testing.T) { + tool, _ := newListTool([]store.SubagentTaskData{ + delegationRow("chat-a", "session-before-reset", "brain", time.Unix(100, 0)), + }) + + // Same chat, a session that did not exist when the delegation was created. + ctx := WithToolSessionKey(listCtx("chat-a"), "session-after-reset") + got := decodeList(t, tool.executeListCompletions(ctx)) + if len(got) != 1 { + t.Fatalf("listed %d entries after a session reset; the delegation is still "+ + "in this chat and must remain recoverable", len(got)) + } +} + +// Without a chat there is nothing to scope to, and listing everything this agent +// ever did would cross the boundary the predicate exists to draw. +func TestDelegateListRefusesWithoutChat(t *testing.T) { + tool, s := newListTool([]store.SubagentTaskData{ + delegationRow("chat-a", "session-1", "brain", time.Unix(100, 0)), + }) + + ctx := store.WithAgentID(store.WithTenantID(context.Background(), uuid.New()), uuid.New()) + res := tool.executeListCompletions(ctx) + if res == nil || !res.IsError { + t.Fatalf("list without a chat returned %#v", res) + } + if s.listCalls != 0 { + t.Errorf("store was queried %d times despite having no chat to scope to", s.listCalls) + } +} + +func TestDelegateListStopsAtItsLimit(t *testing.T) { + rows := make([]store.SubagentTaskData, 0, delegateListLimit+5) + for i := range delegateListLimit + 5 { + rows = append(rows, delegationRow("chat-a", "session-1", fmt.Sprintf("agent-%d", i), time.Unix(int64(i), 0))) + } + tool, _ := newListTool(rows) + + if got := decodeList(t, tool.executeListCompletions(listCtx("chat-a"))); len(got) != delegateListLimit { + t.Fatalf("listed %d entries, want the cap of %d", len(got), delegateListLimit) + } +} diff --git a/internal/tools/delegate_tool.go b/internal/tools/delegate_tool.go index a094fc65..55e5f3f7 100644 --- a/internal/tools/delegate_tool.go +++ b/internal/tools/delegate_tool.go @@ -226,12 +226,12 @@ func (t *DelegateTool) Parameters() map[string]any { }, "action": map[string]any{ "type": "string", - "enum": []string{"delegate", "get"}, - "description": "delegate (default) starts work; get retrieves a durable async result", + "enum": []string{"delegate", "get", "list"}, + "description": "delegate (default) starts work; get retrieves a durable async result by id; list shows the delegations started in this chat, newest first — use it when you no longer have the id", }, "delegation_id": map[string]any{ "type": "string", - "description": "Delegation UUID returned by async mode (required for action=get)", + "description": "Delegation UUID returned by async mode (required for action=get). If you no longer have it, call action=list rather than guessing — a wrong id cannot be recovered from", }, "task": map[string]any{ "type": "string", @@ -264,6 +264,9 @@ func (t *DelegateTool) Execute(ctx context.Context, args map[string]any) *Result if action == "get" { return t.executeGetCompletion(ctx, args) } + if action == "list" { + return t.executeListCompletions(ctx) + } if action != "delegate" { return ErrorResult(fmt.Sprintf("unknown delegate action %q", action)) } diff --git a/internal/tools/subagent_task_lifecycle_test.go b/internal/tools/subagent_task_lifecycle_test.go index d44e7039..3e63520e 100644 --- a/internal/tools/subagent_task_lifecycle_test.go +++ b/internal/tools/subagent_task_lifecycle_test.go @@ -144,6 +144,10 @@ func (s *recordingSubagentTaskStore) ListByParent(context.Context, uuid.UUID, st return nil, nil } +func (s *recordingSubagentTaskStore) ListDelegationsByChat(context.Context, uuid.UUID, string) ([]store.SubagentTaskData, error) { + return nil, nil +} + func (s *recordingSubagentTaskStore) ListBySession(context.Context, uuid.UUID, string) ([]store.SubagentTaskData, error) { return nil, nil }