molecule-ai/molecule-core
41
0
Fork 2
Code Issues 47 Pull Requests 1 Actions 716 Packages Projects Releases Wiki Activity

fix(delegation): write delegation_id into response_body column (mc#984) #998

Merged
devops-engineer merged 8 commits from fix/984-delegation-id-response-body into main 2026-05-14 12:18:22 +00:00
Conversation 6 Commits 8 Files Changed 4 +56 -15
4 changed files with 56 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
workspace-server/internal/handlers/channels_test.go
View File

@ -15,6 +15,7 @@ import (
sqlmock "github.com/DATA-DOG/go-sqlmock"
"github.com/Molecule-AI/molecule-monorepo/platform/internal/channels"
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
"github.com/gin-gonic/gin"
)
@ -364,6 +365,20 @@ func TestChannelHandler_Discover_MissingToken(t *testing.T) {
}
func TestChannelHandler_Discover_UnsupportedType(t *testing.T) {
// Set up db.DB so PausePollersForToken (called inside Discover) doesn't panic.
mockDB, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("sqlmock: %v", err)
}
t.Cleanup(func() { mockDB.Close() })
prevDB := db.DB
db.DB = mockDB
t.Cleanup(func() { db.DB = prevDB })
mock.ExpectQuery(`SELECT id, channel_config FROM workspace_channels WHERE enabled = true AND workspace_id`).
WithArgs("ws-test").
WillReturnRows(sqlmock.NewRows([]string{"id", "channel_config"}))
handler := NewChannelHandler(newTestChannelManager())
// #329: workspace_id required — include so we actually reach the
@ -387,6 +402,20 @@ func TestChannelHandler_Discover_UnsupportedType(t *testing.T) {
}
func TestChannelHandler_Discover_InvalidBotToken(t *testing.T) {
// Set up db.DB so PausePollersForToken (called inside Discover) doesn't panic.
mockDB, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("sqlmock: %v", err)
}
t.Cleanup(func() { mockDB.Close() })
prevDB := db.DB
db.DB = mockDB
t.Cleanup(func() { db.DB = prevDB })
mock.ExpectQuery(`SELECT id, channel_config FROM workspace_channels WHERE enabled = true AND workspace_id`).
WithArgs("ws-test").
WillReturnRows(sqlmock.NewRows([]string{"id", "channel_config"}))
handler := NewChannelHandler(newTestChannelManager())
body, _ := json.Marshal(map[string]interface{}{

23
workspace-server/internal/handlers/delegation.go
View File

@ -262,14 +262,20 @@ func insertDelegationRow(ctx context.Context, c *gin.Context, sourceID string, b
"task": body.Task,
"delegation_id": delegationID,
})
// Store delegation_id in response_body so agent check_delegation_status
// (which reads response_body->>delegation_id) can locate this row even
// when request_body hasn't propagated yet. Fixes mc#984.
respJSON, _ := json.Marshal(map[string]interface{}{
"delegation_id": delegationID,
})
var idemArg interface{}
if body.IdempotencyKey != "" {
idemArg = body.IdempotencyKey
}
_, err := db.DB.ExecContext(ctx, `
INSERT INTO activity_logs (workspace_id, activity_type, method, source_id, target_id, summary, request_body, status, idempotency_key)
VALUES ($1, 'delegation', 'delegate', $2, $3, $4, $5::jsonb, 'pending', $6)
`, sourceID, sourceID, body.TargetID, "Delegating to "+body.TargetID, string(taskJSON), idemArg)
INSERT INTO activity_logs (workspace_id, activity_type, method, source_id, target_id, summary, request_body, response_body, status, idempotency_key)
VALUES ($1, 'delegation', 'delegate', $2, $3, $4, $5::jsonb, $6::jsonb, 'pending', $7)
`, sourceID, sourceID, body.TargetID, "Delegating to "+body.TargetID, string(taskJSON), string(respJSON), idemArg)
if err == nil {
// RFC #2829 #318 — mirror to the durable delegations ledger
// (gated by DELEGATION_LEDGER_WRITE; default off → no-op).
@ -544,10 +550,15 @@ func (h *DelegationHandler) Record(c *gin.Context) {
"task": body.Task,
"delegation_id": body.DelegationID,
})
// Store delegation_id in response_body so agent check_delegation_status
// can locate this row. Fixes mc#984.
respJSON, _ := json.Marshal(map[string]interface{}{
"delegation_id": body.DelegationID,
})
if _, err := db.DB.ExecContext(ctx, `
INSERT INTO activity_logs (workspace_id, activity_type, method, source_id, target_id, summary, request_body, status)
VALUES ($1, 'delegation', 'delegate', $2, $3, $4, $5::jsonb, 'dispatched')
`, sourceID, sourceID, body.TargetID, "Delegating to "+body.TargetID, string(taskJSON)); err != nil {
INSERT INTO activity_logs (workspace_id, activity_type, method, source_id, target_id, summary, request_body, response_body, status)
VALUES ($1, 'delegation', 'delegate', $2, $3, $4, $5::jsonb, $6::jsonb, 'dispatched')
`, sourceID, sourceID, body.TargetID, "Delegating to "+body.TargetID, string(taskJSON), string(respJSON)); err != nil {
log.Printf("Delegation Record: insert failed for %s: %v", body.DelegationID, err)
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to record delegation"})
return

17
workspace-server/internal/handlers/delegation_test.go
View File

@ -133,9 +133,9 @@ func TestDelegate_Success(t *testing.T) {
targetID := "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
// Expect INSERT into activity_logs for delegation tracking
// (6th arg is idempotency_key — nil here since the request omits it)
// (6th arg is response_body, 7th is idempotency_key — nil here since the request omits it)
mock.ExpectExec("INSERT INTO activity_logs").
WithArgs("ws-source", "ws-source", targetID, "Delegating to "+targetID, sqlmock.AnyArg(), nil).
WithArgs("ws-source", "ws-source", targetID, "Delegating to "+targetID, sqlmock.AnyArg(), sqlmock.AnyArg(), nil).
WillReturnResult(sqlmock.NewResult(0, 1))
// Expect RecordAndBroadcast INSERT into structure_events
@ -189,9 +189,9 @@ func TestDelegate_DBInsertFails_Still202WithWarning(t *testing.T) {
targetID := "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
// DB insert fails (6th arg = idempotency_key, nil for this test)
// DB insert fails (6th arg = response_body, 7th = idempotency_key, nil for this test)
mock.ExpectExec("INSERT INTO activity_logs").
WithArgs("ws-source", "ws-source", targetID, "Delegating to "+targetID, sqlmock.AnyArg(), nil).
WithArgs("ws-source", "ws-source", targetID, "Delegating to "+targetID, sqlmock.AnyArg(), sqlmock.AnyArg(), nil).
WillReturnError(fmt.Errorf("database connection lost"))
// RecordAndBroadcast still fires
@ -491,6 +491,7 @@ func TestDelegationRecord_InsertsActivityLogRow(t *testing.T) {
"550e8400-e29b-41d4-a716-446655440001", // target_id
"Delegating to 550e8400-e29b-41d4-a716-446655440001", // summary
sqlmock.AnyArg(), // request_body (jsonb)
sqlmock.AnyArg(), // response_body (jsonb) — mc#984 fix
).
WillReturnResult(sqlmock.NewResult(0, 1))
// RecordAndBroadcast INSERT for DELEGATION_SENT
@ -699,9 +700,9 @@ func TestDelegate_IdempotentFailedRowIsReleasedAndReplaced(t *testing.T) {
mock.ExpectExec("DELETE FROM activity_logs").
WithArgs("ws-source", "retry-key").
WillReturnResult(sqlmock.NewResult(0, 1))
// Fresh insert with the same idempotency key.
// Fresh insert with the same idempotency key (response_body added as mc#984 fix).
mock.ExpectExec("INSERT INTO activity_logs").
WithArgs("ws-source", "ws-source", targetID, "Delegating to "+targetID, sqlmock.AnyArg(), "retry-key").
WithArgs("ws-source", "ws-source", targetID, "Delegating to "+targetID, sqlmock.AnyArg(), sqlmock.AnyArg(), "retry-key").
WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectExec("INSERT INTO structure_events").
WillReturnResult(sqlmock.NewResult(0, 1))
@ -745,9 +746,9 @@ func TestDelegate_IdempotentRaceUniqueViolationReturnsExisting(t *testing.T) {
mock.ExpectQuery("SELECT request_body->>'delegation_id', status, target_id").
WithArgs("ws-source", "race-key").
WillReturnError(fmt.Errorf("sql: no rows in result set"))
// Insert loses the race against a concurrent caller.
// Insert loses the race against a concurrent caller (response_body added as mc#984 fix).
mock.ExpectExec("INSERT INTO activity_logs").
WithArgs("ws-source", "ws-source", targetID, "Delegating to "+targetID, sqlmock.AnyArg(), "race-key").
WithArgs("ws-source", "ws-source", targetID, "Delegating to "+targetID, sqlmock.AnyArg(), sqlmock.AnyArg(), "race-key").
WillReturnError(fmt.Errorf("pq: duplicate key value violates unique constraint \"activity_logs_idempotency_uniq\""))
// Re-query returns the winner.
mock.ExpectQuery("SELECT request_body->>'delegation_id', status").

2
workspace-server/internal/handlers/handlers_test.go
View File

@ -367,7 +367,7 @@ func TestBuildProvisionerConfig_IncludesAwarenessSettings(t *testing.T) {
"ws-123",
"/tmp/configs/template",
map[string][]byte{"config.yaml": []byte("name: test")},
models.CreateWorkspacePayload{Tier: 2, Runtime: "claude-code"},
models.CreateWorkspacePayload{Tier: 2, Runtime: "claude-code", WorkspaceDir: "/tmp/workspace", WorkspaceAccess: "read_write"},
map[string]string{"OPENAI_API_KEY": "sk-test"},
"/tmp/plugins",
"workspace:ws-123",