Agent-System-Backend/internal/runmgr/project_work_gate.go

package runmgr

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"

	"agent-system-backend/internal/store"
	"agent-system-backend/pkg/extension"
	core "agent-system-core"
)

type projectWorkPreflight func(context.Context, string, string) (extension.ProjectWorkDecision, error)
type projectRiskRespond func(context.Context, extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error)
type projectRiskAsk func(core.HITLCard) (core.HITLResponse, error)

var errProjectWorkPending = errors.New("project work source analysis pending")

func (c *runCtx) stageAssistantWorkSources(parts []core.Part) error {
	if !c.mgr.projectWorkObservationEnabled || c.run.ProjectID == "" {
		return nil
	}
	for i, part := range parts {
		var body string
		switch part.Type {
		case core.PartText:
			body = part.Text
		case core.PartPlan, core.PartCheck, core.PartCard:
			raw, err := json.Marshal(part)
			if err != nil {
				return err
			}
			body = string(raw)
		default:
			continue
		}
		if body == "" {
			continue
		}
		hash := sha256.Sum256([]byte(body))
		if _, err := c.mgr.st.AppendRunWorkObservation(store.RunWorkObservationInput{
			RunID: string(c.run.ID), SourceKind: "visible-analysis",
			SourceID:      fmt.Sprintf("%s:assistant-final:%d", c.run.ID, i),
			SourceVersion: hex.EncodeToString(hash[:]), Text: body, Delivery: "complete", PartIndex: i,
			Metadata: map[string]string{"partType": string(part.Type), "stage": "assistant-final"},
		}); err != nil {
			return err
		}
	}
	return nil
}

func resolveProjectWorkDecision(ctx context.Context, actor, projectID, runID, action string,
	preflight projectWorkPreflight, respond projectRiskRespond, ask projectRiskAsk,
) (extension.ProjectWorkDecision, error) {
	pending := extension.ProjectWorkDecision{State: extension.ProjectWorkPending, Reason: "project work risk decision incomplete"}
	if preflight == nil {
		return pending, errors.New("project work gate unavailable")
	}
	decision, err := preflight(ctx, runID, action)
	if err != nil {
		return pending, err
	}
	switch decision.State {
	case extension.ProjectWorkProceed:
		if decision.DecisionID == "" {
			return pending, errors.New("project work proceed decision is not committed")
		}
		return decision, nil
	case extension.ProjectWorkDenied:
		return decision, errors.New("project work action denied: " + decision.Reason)
	case extension.ProjectWorkPending:
		return pending, errProjectWorkPending
	case extension.ProjectWorkClarify:
		if decision.RiskID == "" || decision.RiskRevision == "" || decision.Question == "" || respond == nil || ask == nil {
			return pending, errors.New("project work clarification is incomplete")
		}
	default:
		return pending, fmt.Errorf("invalid project work decision %q", decision.State)
	}
	choices := make([]core.HITLChoice, 0, len(decision.Options))
	for i, option := range decision.Options {
		choices = append(choices, core.HITLChoice{
			ID: fmt.Sprintf("option-%d", i), Label: option,
			Description: "澄清这条项目信息的实际含义", Impact: "选择结果将用于重新判断相关项目动作",
			Recommended: i == 0,
		})
	}
	response, err := ask(core.HITLCard{
		Question: decision.Question, Kind: core.HITLKindChoice, Choices: choices,
		AllowFreeInput: false, Blocked: true,
		BlockReason: decision.Reason,
	})
	if err != nil {
		return pending, err
	}
	answer := response.FreeText
	if response.ChoiceID != "" {
		for _, choice := range choices {
			if choice.ID == response.ChoiceID {
				answer = choice.Label
				break
			}
		}
	}
	if answer == "" {
		return pending, errors.New("project work clarification answer is empty")
	}
	idempotency := sha256.Sum256([]byte(decision.RiskID + "\x00" + decision.RiskRevision + "\x00" + runID + "\x00" + answer))
	resolved, err := respond(ctx, extension.ProjectRiskAnswer{
		Actor: actor, Owner: actor, ProjectID: core.ID(projectID), RiskID: decision.RiskID,
		RiskRevision: decision.RiskRevision, Answer: answer, IdempotencyKey: fmt.Sprintf("%x", idempotency),
	})
	if err != nil {
		return pending, err
	}
	if resolved.State != extension.ProjectWorkProceed || resolved.DecisionID == "" {
		return pending, errors.New("project work clarification did not commit a proceed decision")
	}
	refreshed, err := preflight(ctx, runID, action)
	if err != nil {
		return pending, err
	}
	if refreshed.State != extension.ProjectWorkProceed || refreshed.DecisionID == "" {
		return pending, errors.New("project work sources changed or remain blocked after clarification")
	}
	if decision.SourceWatermark != "" && refreshed.SourceWatermark != decision.SourceWatermark {
		return pending, errors.New("project work source watermark changed while clarification was pending")
	}
	return refreshed, nil
}