goclaw/internal/tools/delegate_completion_ledger.go

package tools

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	"github.com/google/uuid"

	"github.com/nextlevelbuilder/goclaw/internal/store"
)

const delegateCompletionTargetKey = "target_agent_key"

const delegateRunningPersistenceTimeout = 250 * time.Millisecond

func (t *DelegateTool) createDelegateCompletion(ctx context.Context, req DelegateRequest) error {
	if t.taskStore == nil {
		return nil
	}
	delegationID, err := uuid.Parse(req.DelegationID)
	if err != nil {
		return fmt.Errorf("invalid delegation ID: %w", err)
	}
	tenantID := parseUUIDOrNil(req.TenantID)
	if tenantID == uuid.Nil || req.FromAgentID == uuid.Nil {
		return fmt.Errorf("delegation completion requires tenant and source agent")
	}
	var sessionKey, originChannel, originChatID, originPeerKind, originUserID *string
	if req.SessionKey != "" {
		sessionKey = &req.SessionKey
	}
	if req.Channel != "" {
		originChannel = &req.Channel
	}
	if req.ChatID != "" {
		originChatID = &req.ChatID
	}
	if req.PeerKind != "" {
		originPeerKind = &req.PeerKind
	}
	if req.UserID != "" {
		originUserID = &req.UserID
	}
	data := &store.SubagentTaskData{
		BaseModel:      store.BaseModel{ID: delegationID},
		TenantID:       tenantID,
		RootAgentID:    req.FromAgentID,
		ParentAgentKey: req.FromAgentKey,
		SessionKey:     sessionKey,
		Subject:        "Delegate to " + req.ToAgentKey,
		Description:    req.Task,
		Status:         TaskStatusQueued,
		Depth:          subagentDepthFromContext(ctx, 0) + 1,
		OriginChannel:  originChannel,
		OriginChatID:   originChatID,
		OriginPeerKind: originPeerKind,
		OriginUserID:   originUserID,
		Metadata: map[string]any{
			asyncCompletionKindKey:      asyncCompletionKindDelegate,
			delegateCompletionTargetKey: req.ToAgentKey,
			asyncCompletionDeliveryKey:  asyncCompletionDeliveryPending,
		},
	}
	dbCtx := store.WithTenantID(context.Background(), tenantID)
	if err := retryAsyncPersistence(dbCtx, func(attemptCtx context.Context) error {
		return t.taskStore.Create(attemptCtx, data)
	}); err != nil {
		return fmt.Errorf("persist accepted delegation %s: %w", req.DelegationID, err)
	}
	return nil
}

func (t *DelegateTool) updateDelegateCompletion(
	req DelegateRequest,
	status string,
	result *string,
) error {
	if t.taskStore == nil {
		return nil
	}
	delegationID := parseUUIDOrNil(req.DelegationID)
	tenantID := parseUUIDOrNil(req.TenantID)
	if delegationID == uuid.Nil || tenantID == uuid.Nil || req.FromAgentID == uuid.Nil {
		return fmt.Errorf("delegation completion requires valid tenant, source agent, and delegation ID")
	}
	dbCtx := store.WithTenantID(context.Background(), tenantID)
	if err := retryTerminalPersistence(dbCtx, func(attemptCtx context.Context) error {
		return t.taskStore.UpdateStatus(
			attemptCtx,
			req.FromAgentID,
			delegationID,
			status,
			result,
			0,
			0,
			0,
		)
	}); err != nil {
		slog.Error("delegate.async.completion_persist_failed",
			"delegation_id", req.DelegationID,
			"from_agent_id", req.FromAgentID,
			"status", status,
			"error", err,
		)
		return err
	}
	return nil
}

// updateDelegateRunning is observability-only: execution may proceed even when
// this transition cannot be persisted. Keep it to one short attempt outside
// the admission callback so a database outage cannot monopolize a child-run
// permit.
func (t *DelegateTool) updateDelegateRunning(req DelegateRequest) error {
	if t.taskStore == nil {
		return nil
	}
	delegationID := parseUUIDOrNil(req.DelegationID)
	tenantID := parseUUIDOrNil(req.TenantID)
	if delegationID == uuid.Nil || tenantID == uuid.Nil || req.FromAgentID == uuid.Nil {
		return fmt.Errorf("delegation completion requires valid tenant, source agent, and delegation ID")
	}
	dbCtx, cancel := context.WithTimeout(
		store.WithTenantID(context.Background(), tenantID),
		delegateRunningPersistenceTimeout,
	)
	defer cancel()
	err := t.taskStore.UpdateStatus(
		dbCtx,
		req.FromAgentID,
		delegationID,
		TaskStatusRunning,
		nil,
		0,
		0,
		0,
	)
	if err != nil {
		slog.Warn("delegate.async.running_persist_failed",
			"delegation_id", req.DelegationID,
			"from_agent_id", req.FromAgentID,
			"error", err,
		)
	}
	return err
}

func (t *DelegateTool) updateDelegateCompletionMedia(
	req DelegateRequest,
	media []persistedCompletionMedia,
) error {
	if t.taskStore == nil || len(media) == 0 {
		return nil
	}
	delegationID := parseUUIDOrNil(req.DelegationID)
	tenantID := parseUUIDOrNil(req.TenantID)
	if delegationID == uuid.Nil || tenantID == uuid.Nil || req.FromAgentID == uuid.Nil {
		return fmt.Errorf("delegation completion requires valid tenant, source agent, and delegation ID")
	}
	dbCtx := store.WithTenantID(context.Background(), tenantID)
	if err := retryTerminalPersistence(dbCtx, func(attemptCtx context.Context) error {
		return t.taskStore.UpdateMetadata(attemptCtx, req.FromAgentID, delegationID, map[string]any{
			asyncCompletionMediaKey: media,
		})
	}); err != nil {
		slog.Error("delegate.async.completion_media_persist_failed",
			"delegation_id", req.DelegationID,
			"from_agent_id", req.FromAgentID,
			"error", err,
		)
		return err
	}
	return nil
}

func (t *DelegateTool) updateDelegateAnnouncement(req DelegateRequest, delivered bool) {
	if t.taskStore == nil {
		return
	}
	delegationID := parseUUIDOrNil(req.DelegationID)
	tenantID := parseUUIDOrNil(req.TenantID)
	if delegationID == uuid.Nil || tenantID == uuid.Nil || req.FromAgentID == uuid.Nil {
		return
	}
	status := asyncCompletionDeliveryMissed
	if delivered {
		status = asyncCompletionDeliveryDone
	}
	dbCtx := store.WithTenantID(context.Background(), tenantID)
	if err := retryAsyncPersistence(dbCtx, func(attemptCtx context.Context) error {
		return t.taskStore.UpdateMetadata(attemptCtx, req.FromAgentID, delegationID, map[string]any{
			asyncCompletionDeliveryKey: status,
		})
	}); err != nil {
		slog.Warn("delegate.async.announcement_status_failed",
			"delegation_id", req.DelegationID,
			"error", err,
		)
	}
}

func (t *DelegateTool) executeGetCompletion(ctx context.Context, args map[string]any) *Result {
	rawID, _ := args["delegation_id"].(string)
	delegationID, err := uuid.Parse(rawID)
	if err != nil {
		return ErrorResult("delegation_id must be a valid UUID")
	}
	tenantID := store.TenantIDFromContext(ctx)
	fromAgentID := store.AgentIDFromContext(ctx)
	if tenantID == uuid.Nil || fromAgentID == uuid.Nil {
		return ErrorResult("delegate result lookup requires tenant and agent context")
	}
	if t.taskStore == nil {
		return ErrorResult("durable delegation task tracking is unavailable")
	}
	dbCtx := store.WithTenantID(context.WithoutCancel(ctx), tenantID)
	task, err := t.taskStore.Get(dbCtx, fromAgentID, delegationID)
	if err != nil {
		return ErrorResult("failed to load delegation result")
	}
	if task == nil || completionKind(task.Metadata) != asyncCompletionKindDelegate {
		return ErrorResult("delegation result not found")
	}
	payload := persistedCompletionPayload(task)
	payload["delegation_id"] = task.ID.String()
	delete(payload, "completion_id")
	if target, ok := task.Metadata[delegateCompletionTargetKey].(string); ok && target != "" {
		payload["agent"] = target
	}
	encoded, err := json.Marshal(payload)
	if err != nil {
		return ErrorResult("failed to encode delegation result")
	}
	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))
}