molecule-ai/molecule-core
51
0
Fork 2
Code Issues 160 Pull Requests 162 Actions 14 Packages Projects Releases Wiki Activity

fix(handlers): delegation list shows both outgoing and incoming #1362

Open
core-be wants to merge 1 commits from fix/delegation-list-shows-both-directions into main
pull from: fix/delegation-list-shows-both-directions
merge into: molecule-ai:main
Conversation 25 Commits 1 Files Changed 3
+180 -46
3 changed files with 180 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workspace-server/internal/handlers/delegation.go
+18 -8
View File
@@ -680,13 +680,17 @@ func (h *DelegationHandler) ListDelegations(c *gin.Context) {
// listDelegationsFromLedger queries the durable delegations table.
// Returns nil on error so the caller can fall back to activity_logs.
// Includes both outgoing (caller) and incoming (callee) delegations so
// the canvas shows the full delegation history regardless of which side
// the workspace played. A "direction" field distinguishes sent vs. received.
func (h *DelegationHandler) listDelegationsFromLedger(ctx context.Context, workspaceID string) []map[string]interface{} {
rows, err := db.DB.QueryContext(ctx, `
SELECT d.delegation_id, d.caller_id, d.callee_id, d.task_preview,
d.status, d.result_preview, d.error_detail, d.last_heartbeat,
d.deadline, d.created_at, d.updated_at
d.deadline, d.created_at, d.updated_at,
CASE WHEN d.caller_id = $1 THEN 'sent' ELSE 'received' END AS direction
FROM delegations d
WHERE d.caller_id = $1
WHERE d.caller_id = $1 OR d.callee_id = $1
ORDER BY d.created_at DESC
LIMIT 50
`, workspaceID)
@@ -699,13 +703,13 @@ func (h *DelegationHandler) listDelegationsFromLedger(ctx context.Context, works
var result []map[string]interface{}
for rows.Next() {
var delegationID, callerID, calleeID, taskPreview, status string
var delegationID, callerID, calleeID, taskPreview, status, direction string
var resultPreview, errorDetail sql.NullString
var lastHeartbeat, deadline, createdAt, updatedAt *time.Time
if err := rows.Scan(
&delegationID, &callerID, &calleeID, &taskPreview,
&status, &resultPreview, &errorDetail, &lastHeartbeat,
&deadline, &createdAt, &updatedAt,
&deadline, &createdAt, &updatedAt, &direction,
); err != nil {
continue
}
@@ -713,6 +717,7 @@ func (h *DelegationHandler) listDelegationsFromLedger(ctx context.Context, works
"delegation_id": delegationID,
"source_id": callerID,
"target_id": calleeID,
"direction": direction,
"summary": textutil.TruncateBytes(taskPreview, 200),
"status": status,
"created_at": createdAt,
@@ -753,9 +758,9 @@ func (h *DelegationHandler) listDelegationsFromActivityLogs(ctx context.Context,
COALESCE(summary, ''), COALESCE(status, ''), COALESCE(error_detail, ''),
COALESCE(response_body->>'text', response_body::text, ''),
COALESCE(request_body->>'delegation_id', response_body->>'delegation_id', ''),
created_at
created_at, workspace_id
FROM activity_logs
WHERE workspace_id = $1 AND method IN ('delegate', 'delegate_result')
WHERE source_id = $1 AND method IN ('delegate', 'delegate_result')
ORDER BY created_at DESC
LIMIT 50
`, workspaceID)
@@ -766,16 +771,21 @@ func (h *DelegationHandler) listDelegationsFromActivityLogs(ctx context.Context,
var result []map[string]interface{}
for rows.Next() {
var id, actType, sourceID, targetID, summary, status, errorDetail, responseBody, delegationID string
var id, actType, sourceID, targetID, summary, status, errorDetail, responseBody, delegationID, actWorkspaceID string
var createdAt time.Time
if err := rows.Scan(&id, &actType, &sourceID, &targetID, &summary, &status, &errorDetail, &responseBody, &delegationID, &createdAt); err != nil {
if err := rows.Scan(&id, &actType, &sourceID, &targetID, &summary, &status, &errorDetail, &responseBody, &delegationID, &createdAt, &actWorkspaceID); err != nil {
continue
}
direction := "sent"
if actWorkspaceID != sourceID {
direction = "received"
}
entry := map[string]interface{}{
"id": id,
"type": actType,
"source_id": sourceID,
"target_id": targetID,
"direction": direction,
"summary": summary,
"status": status,
"created_at": createdAt,
workspace-server/internal/handlers/delegation_list_test.go
+138 -20
View File
@@ -27,7 +27,7 @@ func TestListDelegationsFromLedger_EmptyResult(t *testing.T) {
rows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail",
"last_heartbeat", "deadline", "created_at", "updated_at",
"last_heartbeat", "deadline", "created_at", "updated_at", "direction",
})
mock.ExpectQuery("SELECT .+ FROM delegations").
WithArgs("ws-1").
@@ -62,11 +62,11 @@ func TestListDelegationsFromLedger_SingleRow(t *testing.T) {
rows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail",
"last_heartbeat", "deadline", "created_at", "updated_at",
"last_heartbeat", "deadline", "created_at", "updated_at", "direction",
}).AddRow(
"del-1", "ws-1", "ws-2", "summarise the report",
"completed", "the report is about Q1",
"", now, now, now, now,
"", now, now, now, now, "sent",
)
mock.ExpectQuery("SELECT .+ FROM delegations").
WithArgs("ws-1").
@@ -102,6 +102,9 @@ func TestListDelegationsFromLedger_SingleRow(t *testing.T) {
if e["_ledger"] != true {
t.Errorf("_ledger marker: got %v, want true", e["_ledger"])
}
if e["direction"] != "sent" {
t.Errorf("direction: got %v, want sent", e["direction"])
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("sqlmock expectations: %v", err)
}
@@ -120,11 +123,11 @@ func TestListDelegationsFromLedger_MultipleRows(t *testing.T) {
rows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail",
"last_heartbeat", "deadline", "created_at", "updated_at",
"last_heartbeat", "deadline", "created_at", "updated_at", "direction",
}).
AddRow("del-a", "ws-1", "ws-2", "task a", "in_progress", "", "", now, now, now, now).
AddRow("del-b", "ws-1", "ws-3", "task b", "failed", "", "timeout", now, now, now, now).
AddRow("del-c", "ws-1", "ws-4", "task c", "completed", "result c", "", now, now, now, now)
AddRow("del-a", "ws-1", "ws-2", "task a", "in_progress", "", "", now, now, now, now, "sent").
AddRow("del-b", "ws-1", "ws-3", "task b", "failed", "", "timeout", now, now, now, now, "sent").
AddRow("del-c", "ws-1", "ws-4", "task c", "completed", "result c", "", now, now, now, now, "sent")
mock.ExpectQuery("SELECT .+ FROM delegations").
WithArgs("ws-1").
WillReturnRows(rows)
@@ -160,9 +163,9 @@ func TestListDelegationsFromLedger_NullsOmitted(t *testing.T) {
rows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail",
"last_heartbeat", "deadline", "created_at", "updated_at",
"last_heartbeat", "deadline", "created_at", "updated_at", "direction",
}).
AddRow("del-1", "ws-1", "ws-2", "task", "queued", nil, nil, nil, nil, now, now)
AddRow("del-1", "ws-1", "ws-2", "task", "queued", nil, nil, nil, nil, now, now, "sent")
mock.ExpectQuery("SELECT .+ FROM delegations").
WithArgs("ws-1").
WillReturnRows(rows)
@@ -239,10 +242,10 @@ func TestListDelegationsFromLedger_RowsErr(t *testing.T) {
rows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail",
"last_heartbeat", "deadline", "created_at", "updated_at",
"last_heartbeat", "deadline", "created_at", "updated_at", "direction",
}).
AddRow("del-1", "ws-1", "ws-2", "task", "queued", "", "", now, now, now, now).
AddRow("del-2", "ws-1", "ws-3", "another task", "queued", "", "", now, now, now, now).
AddRow("del-1", "ws-1", "ws-2", "task", "queued", "", "", now, now, now, now, "sent").
AddRow("del-2", "ws-1", "ws-3", "another task", "queued", "", "", now, now, now, now, "sent").
RowError(1, context.DeadlineExceeded)
mock.ExpectQuery("SELECT .+ FROM delegations").
WithArgs("ws-1").
@@ -287,7 +290,7 @@ func TestListDelegationsFromActivityLogs_EmptyResult(t *testing.T) {
rows := sqlmock.NewRows([]string{
"id", "activity_type", "source_id", "target_id",
"summary", "status", "error_detail",
"response_preview", "delegation_id", "created_at",
"response_preview", "delegation_id", "created_at", "workspace_id",
})
mock.ExpectQuery("SELECT .+ FROM activity_logs").
WithArgs("ws-1").
@@ -319,14 +322,14 @@ func TestListDelegationsFromActivityLogs_SingleDelegateRow(t *testing.T) {
rows := sqlmock.NewRows([]string{
"id", "activity_type", "source_id", "target_id",
"summary", "status", "error_detail",
"response_preview", "delegation_id", "created_at",
"response_preview", "delegation_id", "created_at", "workspace_id",
}).AddRow(
"act-1", "delegate",
"ws-1", "ws-2",
"analyse Q1 numbers",
"in_progress",
"", "", "",
now,
now, "ws-1",
)
mock.ExpectQuery("SELECT .+ FROM activity_logs").
WithArgs("ws-1").
@@ -359,6 +362,9 @@ func TestListDelegationsFromActivityLogs_SingleDelegateRow(t *testing.T) {
if e["status"] != "in_progress" {
t.Errorf("status: got %v", e["status"])
}
if e["direction"] != "sent" {
t.Errorf("direction: got %v, want sent", e["direction"])
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("sqlmock expectations: %v", err)
}
@@ -377,7 +383,7 @@ func TestListDelegationsFromActivityLogs_DelegateResultWithError(t *testing.T) {
rows := sqlmock.NewRows([]string{
"id", "activity_type", "source_id", "target_id",
"summary", "status", "error_detail",
"response_preview", "delegation_id", "created_at",
"response_preview", "delegation_id", "created_at", "workspace_id",
}).AddRow(
"act-2", "delegate_result",
"ws-1", "ws-2",
@@ -386,7 +392,7 @@ func TestListDelegationsFromActivityLogs_DelegateResultWithError(t *testing.T) {
"Callee workspace not reachable",
`{"text":"the result body text"}`,
"del-abc",
now,
now, "ws-1",
)
mock.ExpectQuery("SELECT .+ FROM activity_logs").
WithArgs("ws-1").
@@ -463,10 +469,10 @@ func TestListDelegationsFromActivityLogs_RowsErr(t *testing.T) {
rows := sqlmock.NewRows([]string{
"id", "activity_type", "source_id", "target_id",
"summary", "status", "error_detail",
"response_preview", "delegation_id", "created_at",
"response_preview", "delegation_id", "created_at", "workspace_id",
}).
AddRow("act-1", "delegate", "ws-1", "ws-2", "task", "queued", "", "", "", now).
AddRow("act-2", "delegate", "ws-1", "ws-3", "another task", "queued", "", "", "", now).
AddRow("act-1", "delegate", "ws-1", "ws-2", "task", "queued", "", "", "", now, "ws-1").
AddRow("act-2", "delegate", "ws-1", "ws-3", "another task", "queued", "", "", "", now, "ws-1").
RowError(1, context.DeadlineExceeded)
mock.ExpectQuery("SELECT .+ FROM activity_logs").
WithArgs("ws-1").
@@ -493,3 +499,115 @@ func TestListDelegationsFromActivityLogs_RowsErr(t *testing.T) {
// sqlmock.NewRows([]string{}).AddRow(...) to panic in test SETUP. The handler
// has no recover(), so a scan panic would crash the process — the correct
// behaviour. Real-DB integration tests cover this path.
// ---------- Direction: received (callee) ----------
func TestListDelegationsFromLedger_CalleeDirection_Received(t *testing.T) {
// When the workspace ID appears as callee_id (not caller_id), direction = "received".
// The query returns rows where d.caller_id = $1 OR d.callee_id = $1, and the
// CASE expression sets direction based on whether caller_id matches.
mockDB, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("failed to create sqlmock: %v", err)
}
prevDB := db.DB
db.DB = mockDB
t.Cleanup(func() { db.DB = prevDB; mockDB.Close() })
now := time.Now()
// ws-1 is the callee here (received a delegation from ws-other)
rows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail",
"last_heartbeat", "deadline", "created_at", "updated_at", "direction",
}).AddRow(
"del-received-1", "ws-other", "ws-1", "task from other workspace",
"in_progress", "", "",
now, now, now, now, "received",
)
mock.ExpectQuery("SELECT .+ FROM delegations").
WithArgs("ws-1").
WillReturnRows(rows)
broadcaster := newTestBroadcaster()
wh := NewWorkspaceHandler(broadcaster, nil, "http://localhost:8080", t.TempDir())
dh := NewDelegationHandler(wh, broadcaster)
got := dh.listDelegationsFromLedger(context.Background(), "ws-1")
if len(got) != 1 {
t.Fatalf("expected 1 entry, got %d", len(got))
}
e := got[0]
if e["delegation_id"] != "del-received-1" {
t.Errorf("delegation_id: got %v, want del-received-1", e["delegation_id"])
}
// source_id is the workspace that initiated the delegation (caller)
if e["source_id"] != "ws-other" {
t.Errorf("source_id: got %v, want ws-other", e["source_id"])
}
// target_id is the workspace receiving the delegation (callee)
if e["target_id"] != "ws-1" {
t.Errorf("target_id: got %v, want ws-1", e["target_id"])
}
if e["direction"] != "received" {
t.Errorf("direction: got %v, want received", e["direction"])
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("sqlmock expectations: %v", err)
}
}
func TestListDelegationsFromActivityLogs_ReceivedDirection(t *testing.T) {
// When workspace_id differs from source_id, direction = "received".
// This happens when the workspace received a delegation, not sent one.
mockDB, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("failed to create sqlmock: %v", err)
}
prevDB := db.DB
db.DB = mockDB
t.Cleanup(func() { db.DB = prevDB; mockDB.Close() })
now := time.Now()
// ws-1 is receiving a delegation from ws-other (workspace_id != source_id)
rows := sqlmock.NewRows([]string{
"id", "activity_type", "source_id", "target_id",
"summary", "status", "error_detail",
"response_preview", "delegation_id", "created_at", "workspace_id",
}).AddRow(
"act-received-1", "delegate",
"ws-other", "ws-1",
"Delegating to ws-1",
"in_progress",
"", "", "",
now, "ws-1", // workspace_id = ws-1 (the receiving workspace)
)
mock.ExpectQuery("SELECT .+ FROM activity_logs").
WithArgs("ws-1").
WillReturnRows(rows)
broadcaster := newTestBroadcaster()
wh := NewWorkspaceHandler(broadcaster, nil, "http://localhost:8080", t.TempDir())
dh := NewDelegationHandler(wh, broadcaster)
got := dh.listDelegationsFromActivityLogs(context.Background(), "ws-1")
if len(got) != 1 {
t.Fatalf("expected 1 entry, got %d", len(got))
}
e := got[0]
if e["id"] != "act-received-1" {
t.Errorf("id: got %v, want act-received-1", e["id"])
}
if e["source_id"] != "ws-other" {
t.Errorf("source_id: got %v, want ws-other", e["source_id"])
}
if e["target_id"] != "ws-1" {
t.Errorf("target_id: got %v, want ws-1", e["target_id"])
}
if e["direction"] != "received" {
t.Errorf("direction: got %v, want received", e["direction"])
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("sqlmock expectations: %v", err)
}
}
workspace-server/internal/handlers/delegation_test.go
+24 -18
View File
@@ -239,14 +239,14 @@ func TestListDelegations_Empty(t *testing.T) {
WillReturnRows(sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail", "last_heartbeat",
"deadline", "created_at", "updated_at",
"deadline", "created_at", "updated_at", "direction",
}))
mock.ExpectQuery("SELECT id, activity_type").
WithArgs("ws-source").
WillReturnRows(sqlmock.NewRows([]string{
"id", "activity_type", "source_id", "target_id",
"summary", "status", "error_detail", "response_body",
"delegation_id", "created_at",
"delegation_id", "created_at", "workspace_id",
}))
w := httptest.NewRecorder()
@@ -287,14 +287,14 @@ func TestListDelegations_WithResults(t *testing.T) {
rows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail", "last_heartbeat",
"deadline", "created_at", "updated_at",
"deadline", "created_at", "updated_at", "direction",
}).
AddRow("del-111", "ws-source", "ws-target",
"Delegating to ws-target", "pending", "", "",
&now, &deadline, now, now).
&now, &deadline, now, now, "sent").
AddRow("del-222", "ws-source", "ws-target",
"Delegation completed (hello world)", "completed", "hello world", "",
&now, &deadline, now, now.Add(time.Minute))
&now, &deadline, now, now.Add(time.Minute), "sent")
mock.ExpectQuery("SELECT d.delegation_id, d.caller_id, d.callee_id, d.task_preview").
WithArgs("ws-source").
@@ -335,6 +335,9 @@ func TestListDelegations_WithResults(t *testing.T) {
if resp[0]["_ledger"] != true {
t.Errorf("expected _ledger=true marker, got %v", resp[0]["_ledger"])
}
if resp[0]["direction"] != "sent" {
t.Errorf("expected direction 'sent', got %v", resp[0]["direction"])
}
// Check second entry (completed, has response_preview)
if resp[1]["delegation_id"] != "del-222" {
@@ -1380,11 +1383,11 @@ func TestListDelegations_LedgerRowsReturned(t *testing.T) {
ledgerRows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail", "last_heartbeat",
"deadline", "created_at", "updated_at",
"deadline", "created_at", "updated_at", "direction",
}).AddRow(
"del-ledger-001", "caller-uuid", "callee-uuid",
"Analyze the codebase for bugs", "in_progress", "", "",
&now, &deadline, now, now,
&now, &deadline, now, now, "sent",
)
mock.ExpectQuery("SELECT d.delegation_id, d.caller_id, d.callee_id, d.task_preview").
WithArgs("caller-uuid").
@@ -1422,6 +1425,9 @@ func TestListDelegations_LedgerRowsReturned(t *testing.T) {
if resp[0]["target_id"] != "callee-uuid" {
t.Errorf("expected target_id 'callee-uuid', got %v", resp[0]["target_id"])
}
if resp[0]["direction"] != "sent" {
t.Errorf("expected direction 'sent', got %v", resp[0]["direction"])
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Errorf("unmet sqlmock expectations: %v", err)
}
@@ -1442,18 +1448,18 @@ func TestListDelegations_LedgerEmptyFallsBackToActivityLogs(t *testing.T) {
WillReturnRows(sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail", "last_heartbeat",
"deadline", "created_at", "updated_at",
"deadline", "created_at", "updated_at", "direction",
}))
now := time.Now()
activityRows := sqlmock.NewRows([]string{
"id", "activity_type", "source_id", "target_id",
"summary", "status", "error_detail", "response_body",
"delegation_id", "created_at",
"delegation_id", "created_at", "workspace_id",
}).AddRow(
"act-001", "delegation", "ws-source", "ws-target",
"Delegating to ws-target", "pending", "", "",
"del-old-001", now,
"del-old-001", now, "ws-source",
)
mock.ExpectQuery("SELECT id, activity_type").
WithArgs("ws-source").
@@ -1502,7 +1508,7 @@ func TestListDelegations_BothEmptyReturnsEmptyArray(t *testing.T) {
WillReturnRows(sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail", "last_heartbeat",
"deadline", "created_at", "updated_at",
"deadline", "created_at", "updated_at", "direction",
}))
// activity_logs also empty
mock.ExpectQuery("SELECT id, activity_type").
@@ -1510,7 +1516,7 @@ func TestListDelegations_BothEmptyReturnsEmptyArray(t *testing.T) {
WillReturnRows(sqlmock.NewRows([]string{
"id", "activity_type", "source_id", "target_id",
"summary", "status", "error_detail", "response_body",
"delegation_id", "created_at",
"delegation_id", "created_at", "workspace_id",
}))
w := httptest.NewRecorder()
@@ -1553,11 +1559,11 @@ func TestListDelegations_LedgerQueryErrorFallsBackToActivityLogs(t *testing.T) {
activityRows := sqlmock.NewRows([]string{
"id", "activity_type", "source_id", "target_id",
"summary", "status", "error_detail", "response_body",
"delegation_id", "created_at",
"delegation_id", "created_at", "workspace_id",
}).AddRow(
"act-002", "delegation", "ws-source", "ws-target",
"Some task", "completed", "", "result here",
"del-pre-318", now,
"del-pre-318", now, "ws-source",
)
mock.ExpectQuery("SELECT id, activity_type").
WithArgs("ws-source").
@@ -1599,11 +1605,11 @@ func TestListDelegations_LedgerCompletedIncludesResultPreview(t *testing.T) {
ledgerRows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail", "last_heartbeat",
"deadline", "created_at", "updated_at",
"deadline", "created_at", "updated_at", "direction",
}).AddRow(
"del-complete-001", "caller-uuid", "callee-uuid",
"Run analysis", "completed", "Analysis complete: 42 issues found", "",
&now, &deadline, now, now,
&now, &deadline, now, now, "sent",
)
mock.ExpectQuery("SELECT d.delegation_id, d.caller_id, d.callee_id, d.task_preview").
WithArgs("caller-uuid").
@@ -1654,11 +1660,11 @@ func TestListDelegations_LedgerFailedIncludesErrorDetail(t *testing.T) {
ledgerRows := sqlmock.NewRows([]string{
"delegation_id", "caller_id", "callee_id", "task_preview",
"status", "result_preview", "error_detail", "last_heartbeat",
"deadline", "created_at", "updated_at",
"deadline", "created_at", "updated_at", "direction",
}).AddRow(
"del-failed-001", "caller-uuid", "callee-uuid",
"Fetch data", "failed", "", "Callee workspace not reachable",
&now, &deadline, now, now,
&now, &deadline, now, now, "sent",
)
mock.ExpectQuery("SELECT d.delegation_id, d.caller_id, d.callee_id, d.task_preview").
WithArgs("caller-uuid").