multica/server/internal/handler/comment_root_projection_test.go

package handler

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/multica-ai/multica/server/internal/testutil"
	db "github.com/multica-ai/multica/server/pkg/db/generated"
)

// Observe actual result columns/rows at the database boundary. A query that
// filters later in Go still transfers the unwanted task payloads and fails.
type rootOwnerReadDB struct {
	db.DBTX
	rows, reads int
	payload     bool
	fail        bool
}
type rootOwnerRows struct {
	pgx.Rows
	observer *rootOwnerReadDB
}

func (r *rootOwnerRows) Next() bool {
	ok := r.Rows.Next()
	if ok {
		r.observer.rows++
	}
	return ok
}
func (d *rootOwnerReadDB) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) {
	if !strings.Contains(sql, "FROM agent_task_queue") {
		return d.DBTX.Query(ctx, sql, args...)
	}
	d.reads++
	if d.fail {
		return nil, errors.New("injected owner read failure")
	}
	rows, err := d.DBTX.Query(ctx, sql, args...)
	if err != nil {
		return nil, err
	}
	for _, field := range rows.FieldDescriptions() {
		if field.Name == "context" || field.Name == "result" {
			d.payload = true
		}
	}
	return &rootOwnerRows{Rows: rows, observer: d}, nil
}

func TestConversationRootOwnersReadOnlyRoutingData(t *testing.T) {
	if testHandler == nil {
		t.Skip("database not available")
	}
	ctx := context.Background()
	agentA := createHandlerTestAgent(t, "Root owner A", nil)
	agentB := createHandlerTestAgent(t, "Root owner B", nil)
	otherAgent := createHandlerTestAgent(t, "Other root owner", nil)
	issueID := dbfx.Issue(t, "Routing projection")
	otherIssueID := dbfx.Issue(t, "Other issue")
	rootID := dbfx.Comment(t, issueID, "No explicit owner")
	otherRootID := dbfx.Comment(t, issueID, "Another root")
	oldSquad := dbfx.Squad(t, "Old root squad", agentA)
	newSquad := dbfx.Squad(t, "New root squad", agentA)
	for i := 0; i < 96; i++ {
		squad := any(nil)
		if i == 0 {
			squad = oldSquad
		}
		if i == 1 {
			squad = newSquad
		}
		dbfx.Task(t, agentA, testutil.Cols{
			"issue_id": issueID, "trigger_comment_id": rootID, "runtime_id": handlerTestRuntimeID(t), "status": "completed", "squad_id": squad,
			"created_at": time.Date(2026, 1, 1, 0, 0, i, 0, time.UTC),
			"context":    testutil.Raw(`jsonb_build_object('large',repeat('x',65536))`),
			"result":     testutil.Raw(`jsonb_build_object('large',repeat('y',65536))`),
		})
	}
	dbfx.Task(t, agentB, testutil.Cols{"issue_id": issueID, "trigger_comment_id": rootID, "runtime_id": handlerTestRuntimeID(t), "status": "failed"})
	dbfx.Task(t, otherAgent, testutil.Cols{"issue_id": issueID, "trigger_comment_id": otherRootID, "runtime_id": handlerTestRuntimeID(t), "status": "completed"})
	dbfx.Task(t, otherAgent, testutil.Cols{"issue_id": otherIssueID, "trigger_comment_id": rootID, "runtime_id": handlerTestRuntimeID(t), "status": "completed"})
	issue, err := testHandler.Queries.GetIssue(ctx, parseUUID(issueID))
	if err != nil {
		t.Fatal(err)
	}
	root, err := testHandler.Queries.GetComment(ctx, parseUUID(rootID))
	if err != nil {
		t.Fatal(err)
	}
	observer := &rootOwnerReadDB{DBTX: testPool}
	h := *testHandler
	h.Queries = db.New(observer)
	triggers, owned := h.routeConversationOwnersForRoot(ctx, issue, root, testUserID, commentTriggerComputeOptions{})
	if !owned || len(triggers) != 2 {
		t.Fatalf("owned=%t triggers=%d, want two root owners", owned, len(triggers))
	}
	for _, trigger := range triggers {
		switch uuidToString(trigger.Agent.ID) {
		case agentA:
			if trigger.Squad == nil || uuidToString(trigger.Squad.ID) != newSquad {
				t.Fatal("lost newest non-null squad")
			}
		case agentB:
			if trigger.Squad != nil {
				t.Fatal("invented squad")
			}
		default:
			t.Fatal("routed unrelated task owner")
		}
	}
	if observer.reads != 1 || observer.rows != 2 || observer.payload {
		t.Fatalf("routing read: reads=%d rows=%d payload=%t; want one narrow row per owner", observer.reads, observer.rows, observer.payload)
	}

	for _, excluded := range []string{rootID, otherRootID} {
		got, owned := h.routeConversationOwnersForRoot(ctx, issue, root, testUserID, commentTriggerComputeOptions{ExcludeTriggerCommentID: parseUUID(excluded)})
		if excluded == rootID {
			if owned || len(got) != 0 {
				t.Fatal("excluded root routed")
			}
		} else if !owned || len(got) != 2 {
			t.Fatal("unrelated exclusion removed owners")
		}
	}
	emptyID := dbfx.Comment(t, issueID, "No routed task")
	empty, err := testHandler.Queries.GetComment(ctx, parseUUID(emptyID))
	if err != nil {
		t.Fatal(err)
	}
	if got, owned := h.routeConversationOwnersForRoot(ctx, issue, empty, testUserID, commentTriggerComputeOptions{}); owned || len(got) != 0 {
		t.Fatal("invented empty root owner")
	}
	observer.fail = true
	if got, owned := h.routeConversationOwnersForRoot(ctx, issue, root, testUserID, commentTriggerComputeOptions{}); owned || len(got) != 0 {
		t.Fatal("read failure routed an owner")
	}
	// An explicit mention remains authoritative, even if historical lookup fails.
	root.Content = fmt.Sprintf("[@Other](mention://agent/%s)", otherAgent)
	if got, owned := h.routeConversationOwnersForRoot(ctx, issue, root, testUserID, commentTriggerComputeOptions{}); !owned || len(got) != 1 || uuidToString(got[0].Agent.ID) != otherAgent {
		t.Fatal("explicit owner lost precedence")
	}
	observer.fail = false
	root.Content = "No explicit owner"
	dbfx.Exec(t, "UPDATE agent SET archived_at=now() WHERE id=$1", agentB)
	if got, owned := h.routeConversationOwnersForRoot(ctx, issue, root, testUserID, commentTriggerComputeOptions{}); !owned || len(got) != 1 || uuidToString(got[0].Agent.ID) != agentA {
		t.Fatal("archived owner remained invokable")
	}
	dbfx.Exec(t, "UPDATE agent SET archived_at=now() WHERE id=$1", agentA)
	if got, owned := h.routeConversationOwnersForRoot(ctx, issue, root, testUserID, commentTriggerComputeOptions{}); !owned || len(got) != 0 {
		t.Fatal("unavailable historical owners must still suppress reassignment fallback")
	}

}