Agent-System/internal/app/run_control_owner.go

package app

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"sync"

	"chat-agent-system/internal/app/intentflow"
	"chat-agent-system/internal/bizdev"
	"chat-agent-system/internal/projectworkflow"
	"chat-agent-system/internal/taskrun"
)

func (r *RunControlOwner) isLiveKey(key string) bool {
	return intentflow.IsRunControlLiveKey(key)
}

// RunControlOwner is the U12 live adapter for workflow.run and run.resume.
type RunControlOwner struct {
	Access      ProjectAccess
	Workflow    *projectworkflow.Service
	TaskRuns    *taskrun.Manager
	ProjectName func(ctx context.Context, projectID int64) (string, error)
	Denied      bool
	dryRun      *intentflow.DryRunAdapter

	mu       sync.Mutex
	receipts map[string]intentflow.Receipt
}

func NewRunControlOwner(access ProjectAccess, pw *projectworkflow.Service, runs *taskrun.Manager) *RunControlOwner {
	return &RunControlOwner{
		Access:   access,
		Workflow: pw,
		TaskRuns: runs,
		dryRun:   intentflow.NewRunControlDryRunAdapter(),
		receipts: map[string]intentflow.Receipt{},
	}
}

func (r *RunControlOwner) Authorize(ctx context.Context, exec intentflow.StepExec) error {
	if r == nil || r.Denied {
		return intentflow.ErrPlanActionForbidden
	}
	if !r.isLiveKey(exec.ActionKey) {
		return r.dryRun.Authorize(ctx, exec)
	}
	if exec.Principal.UserID <= 0 || exec.Principal.ProjectID <= 0 || exec.Principal.SessionID == "" {
		return intentflow.ErrPlanActionForbidden
	}
	if r.Access != nil {
		if err := r.Access(ctx, exec.Principal.ProjectID, exec.Principal.UserID); err != nil {
			return intentflow.ErrPlanActionForbidden
		}
	}
	if exec.ActionKey == "run.resume" || exec.ActionKey == "run.restart" {
		workDir := slotValue(exec, "work_dir")
		if workDir == "" {
			workDir = strings.TrimSpace(exec.SafeSummary)
		}
		if workDir == "" {
			return intentflow.ErrPlanActionForbidden
		}
	}
	if exec.ActionKey == "workflow.switch" {
		if slotValue(exec, "workflow_key") == "" {
			return intentflow.ErrPlanActionForbidden
		}
	}
	return nil
}

func (r *RunControlOwner) Execute(ctx context.Context, exec intentflow.StepExec) (intentflow.Receipt, error) {
	if !r.isLiveKey(exec.ActionKey) {
		return r.dryRun.Execute(ctx, exec)
	}
	if err := r.Authorize(ctx, exec); err != nil {
		return intentflow.Receipt{}, err
	}
	r.mu.Lock()
	if rec, ok := r.receipts[exec.IdempotencyKey]; ok {
		r.mu.Unlock()
		if rec.ActionDigest != exec.ActionDigest {
			return intentflow.Receipt{}, intentflow.ErrPlanDecisionConflict
		}
		return rec, nil
	}
	r.mu.Unlock()

	ref, err := r.runLive(ctx, exec)
	if err != nil {
		return intentflow.Receipt{}, err
	}
	rec := intentflow.Receipt{
		IdempotencyKey: exec.IdempotencyKey,
		ActionDigest:   exec.ActionDigest,
		ResultRef:      ref,
		ResultDigest:   fmt.Sprintf("run_control:%s", exec.ActionKey),
		SafeSummary:    exec.ActionKey + " accepted",
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	if existing, ok := r.receipts[exec.IdempotencyKey]; ok {
		if existing.ActionDigest != exec.ActionDigest {
			return intentflow.Receipt{}, intentflow.ErrPlanDecisionConflict
		}
		return existing, nil
	}
	r.receipts[exec.IdempotencyKey] = rec
	return rec, nil
}

func (r *RunControlOwner) runLive(ctx context.Context, exec intentflow.StepExec) (string, error) {
	switch exec.ActionKey {
	case "workflow.run":
		if r.Workflow == nil {
			return "", fmt.Errorf("run control: workflow service missing")
		}
		workflowKey := slotValue(exec, "workflow_key")
		if workflowKey == "" {
			workflowKey = "main"
		}
		runID := planRunID(exec.IdempotencyKey)
		startedBy := exec.Principal.UserID
		created, err := r.Workflow.CreateWorkflowRun(ctx, &projectworkflow.WorkflowRun{
			RunID:       runID,
			ProjectID:   exec.Principal.ProjectID,
			WorkflowKey: workflowKey,
			StartedBy:   &startedBy,
			Status:      "running",
		})
		if err != nil {
			return "", err
		}
		return "workflow_run:" + created.RunID, nil
	case "run.resume":
		if r.TaskRuns == nil {
			return "", fmt.Errorf("run control: taskrun manager missing")
		}
		workDir := slotValue(exec, "work_dir")
		if workDir == "" {
			workDir = strings.TrimSpace(exec.SafeSummary)
		}
		if workDir == "" {
			return "", intentflow.ErrPlanActionForbidden
		}
		if existing, err := r.TaskRuns.FindByIdempotencyKey(ctx, exec.IdempotencyKey); err != nil {
			return "", err
		} else if existing != nil && existing.RunID != "" {
			return existing.RunID, nil
		}
		projectName := ""
		if r.ProjectName != nil {
			projectName, _ = r.ProjectName(ctx, exec.Principal.ProjectID)
		}
		runID := planRunID(exec.IdempotencyKey)
		rec := &taskrun.RunRecord{
			RunID:          runID,
			SessionID:      exec.Principal.SessionID,
			ProjectID:      exec.Principal.ProjectID,
			UserID:         exec.Principal.UserID,
			IdempotencyKey: exec.IdempotencyKey,
			TaskKey:        "intake",
			Status:         "pending",
		}
		created, err := r.TaskRuns.CreateAndStart(ctx, runID, rec, bizdev.TaskRunRequest{
			TaskKey:     "intake",
			UserID:      exec.Principal.UserID,
			ProjectID:   exec.Principal.ProjectID,
			SessionID:   exec.Principal.SessionID,
			ProjectName: projectName,
		}, bizdev.StageRunRequest{}, workDir)
		if err != nil {
			return "", err
		}
		return created.RunID, nil
	case "run.restart":
		// 重启语义：不续用原 workspace，从 intake 重新起一个全新 run（幂等键防重复）。
		if r.TaskRuns == nil {
			return "", fmt.Errorf("run control: taskrun manager missing")
		}
		if existing, err := r.TaskRuns.FindByIdempotencyKey(ctx, exec.IdempotencyKey); err != nil {
			return "", err
		} else if existing != nil && existing.RunID != "" {
			return existing.RunID, nil
		}
		projectName := ""
		if r.ProjectName != nil {
			projectName, _ = r.ProjectName(ctx, exec.Principal.ProjectID)
		}
		runID := planRunID(exec.IdempotencyKey)
		taskKey := slotValue(exec, "task_key")
		if taskKey == "" {
			taskKey = "intake"
		}
		rec := &taskrun.RunRecord{
			RunID:          runID,
			SessionID:      exec.Principal.SessionID,
			ProjectID:      exec.Principal.ProjectID,
			UserID:         exec.Principal.UserID,
			IdempotencyKey: exec.IdempotencyKey,
			TaskKey:        taskKey,
			Status:         "pending",
		}
		created, err := r.TaskRuns.CreateAndStart(ctx, runID, rec, bizdev.TaskRunRequest{
			TaskKey:     taskKey,
			UserID:      exec.Principal.UserID,
			ProjectID:   exec.Principal.ProjectID,
			SessionID:   exec.Principal.SessionID,
			ProjectName: projectName,
		}, bizdev.StageRunRequest{}, "")
		if err != nil {
			return "", err
		}
		return created.RunID, nil
	case "workflow.switch":
		if r.Workflow == nil {
			return "", fmt.Errorf("run control: workflow service missing")
		}
		workflowKey := slotValue(exec, "workflow_key")
		if workflowKey == "" {
			return "", intentflow.ErrPlanActionForbidden
		}
		res, err := r.Workflow.ActivateWorkflow(ctx, exec.Principal.ProjectID, exec.Principal.UserID,
			projectworkflow.ActivateWorkflowInput{
				WorkflowKey: workflowKey,
				Reason:      slotValue(exec, "reason"),
			})
		if err != nil {
			return "", err
		}
		return fmt.Sprintf("workflow_switch:%d:%s", res.ProjectID, res.ActiveWorkflowKey), nil
	default:
		return "", intentflow.ErrPlanActionForbidden
	}
}

func (r *RunControlOwner) LookupReceipt(ctx context.Context, idempotencyKey string) (intentflow.Receipt, bool, error) {
	if r == nil {
		return intentflow.Receipt{}, false, nil
	}
	r.mu.Lock()
	rec, ok := r.receipts[idempotencyKey]
	r.mu.Unlock()
	if ok {
		return rec, true, nil
	}
	return r.dryRun.LookupReceipt(ctx, idempotencyKey)
}

func slotValue(exec intentflow.StepExec, key string) string {
	if len(exec.Slots) == 0 {
		return ""
	}
	var slots map[string]string
	if json.Unmarshal(exec.Slots, &slots) != nil {
		return ""
	}
	return strings.TrimSpace(slots[key])
}