Agent-System-Backend/internal/wfcompile/project_sources.go

package wfcompile

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sync"
	"time"

	"agent-squad/squad"
)

// verifyPriorWorkflowOutputs is shared by model and host steps. The engine's
// task brief/blackboard/handoff can carry earlier output as a fresh user
// prompt, so every completed machine step must have a current source proof.
func verifyPriorWorkflowOutputs(ctx context.Context, rc ProjectStepProvenance, wsDir string, subOf, kinds map[string]string, modelSend bool) error {
	state, err := (squad.FileStore{}).LoadRun(wsDir)
	if err != nil {
		return fmt.Errorf("project workflow state unavailable: %w", err)
	}
	for _, step := range state.Plan.Workflow.Steps {
		if step.Status != "completed" {
			continue
		}
		switch kinds[step.ID] {
		case "prompt":
			sub := subOf[step.ID]
			if sub == "" || step.AttemptID == "" {
				return fmt.Errorf("project workflow step %s lacks source identity", step.ID)
			}
			data, err := os.ReadFile(filepath.Join(wsDir, filepath.FromSlash(sub), "output.md"))
			if err != nil {
				return fmt.Errorf("project workflow step %s output unavailable: %w", step.ID, err)
			}
			if err := rc.VerifyProjectStepOutput(ctx, step.ID, step.AttemptID, "prompt", data); err != nil {
				return fmt.Errorf("project workflow step %s output source changed: %w", step.ID, err)
			}
		case "hitl", "critic": // human input or a verdict whose prompt source was checked above
		default:
			if modelSend {
				return fmt.Errorf("project workflow step %s has unproven machine output", step.ID)
			}
		}
	}
	return nil
}

// completedStepAttempt comes from the squad engine's persisted state, not a
// model response or plugin argument. It identifies the exact execution that
// produced output.md, including after a workflow resumes.
func completedStepAttempt(wsDir, stepID string) (string, error) {
	state, err := (squad.FileStore{}).LoadRun(wsDir)
	if err != nil {
		return "", err
	}
	for _, step := range state.Plan.Workflow.Steps {
		if step.ID == stepID && step.Status == "completed" && step.AttemptID != "" {
			return step.AttemptID, nil
		}
	}
	return "", errors.New("completed step attempt unavailable")
}

// projectOutputSink binds only a completed prompt step's materialized bytes.
// Parallel prompt executions have no unambiguous request-to-step attribution
// in the current backend contract, so they receive no binding and the critic
// fails closed. A recovered completion without its start event likewise needs
// an existing persisted binding to pass its critic.
type projectOutputSink struct {
	rc        ProjectStepProvenance
	ctx       context.Context
	root      string
	subOf     map[string]string
	kinds     map[string]string
	mu        sync.Mutex
	started   map[string]time.Time
	ambiguous map[string]bool
}

func newProjectOutputSink(rc Host, root string, subOf, kinds map[string]string) squad.ProgressSink {
	binder := rc.(ProjectStepProvenance)
	return &projectOutputSink{rc: binder, ctx: rc.Context(), root: root, subOf: subOf, kinds: kinds,
		started: map[string]time.Time{}, ambiguous: map[string]bool{}}
}

func (s *projectOutputSink) Emit(ev squad.Event) {
	if ev.StepID == "" {
		return
	}
	s.mu.Lock()
	switch ev.Type {
	case squad.EvStepStarted:
		if s.kinds[ev.StepID] == "prompt" {
			if err := s.rc.InvalidateProjectStepOutput(s.ctx, ev.StepID); err != nil {
				s.ambiguous[ev.StepID] = true
			}
		}
		if len(s.started) != 0 {
			if s.kinds[ev.StepID] == "prompt" {
				s.ambiguous[ev.StepID] = true
			}
			for id := range s.started {
				if s.kinds[id] == "prompt" {
					s.ambiguous[id] = true
				}
			}
		}
		s.started[ev.StepID] = ev.At
		s.mu.Unlock()
		return
	case squad.EvStepCompleted:
		startedAt, started := s.started[ev.StepID]
		ambiguous := s.ambiguous[ev.StepID]
		delete(s.started, ev.StepID)
		delete(s.ambiguous, ev.StepID)
		s.mu.Unlock()
		if s.kinds[ev.StepID] != "prompt" || !started || startedAt.IsZero() || ambiguous {
			return
		}
		sub := s.subOf[ev.StepID]
		if sub == "" {
			return
		}
		data, err := os.ReadFile(filepath.Join(s.root, "runs", ev.RunID, filepath.FromSlash(sub), "output.md"))
		if err == nil {
			wsDir := filepath.Join(s.root, "runs", ev.RunID)
			attemptID, attemptErr := completedStepAttempt(wsDir, ev.StepID)
			if attemptErr == nil {
				_ = s.rc.BindProjectStepOutput(s.ctx, ev.StepID, attemptID, "prompt", startedAt, data)
			}
		}
		return
	case squad.EvStepFailed:
		delete(s.started, ev.StepID)
		delete(s.ambiguous, ev.StepID)
		s.mu.Unlock()
		return
	default:
		s.mu.Unlock()
	}
}