iop/apps/agent/internal/taskloop/module.go

package taskloop

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"go.uber.org/zap"
	"gopkg.in/yaml.v3"

	"iop/apps/agent/internal/projectlog"
	"iop/packages/go/agentconfig"
	"iop/packages/go/agentguard"
	"iop/packages/go/agentpolicy"
	"iop/packages/go/agentprovider/catalog"
	clistatus "iop/packages/go/agentprovider/cli/status"
	"iop/packages/go/agentruntime"
	"iop/packages/go/agentstate"
	"iop/packages/go/agenttask"
	"iop/packages/go/agentworkspace"
)

type Config struct {
	Snapshot       agentconfig.RuntimeSnapshot
	Catalog        agentconfig.Catalog
	StateStore     *agentstate.Store
	OwnerID        string
	EventSink      agenttask.EventSink
	ReviewExecutor ReviewExecutor
	Provider       agenttask.ProviderInvoker
	Validator      agentworkspace.ValidationFunc
	QuotaObserver  QuotaObserver
}

type QuotaObserver interface {
	ObserveQuota(
		context.Context,
		agenttask.ExecutionTarget,
		time.Time,
	) (agentpolicy.QuotaObservation, error)
}

// Runtime is the standalone host-facing owner around the one shared manager.
// Read projections load the same durable manager state used by CLI, daemon,
// local control, and project logs.
type Runtime struct {
	snapshot      agentconfig.RuntimeSnapshot
	catalog       agentconfig.Catalog
	state         *agentstate.Store
	selections    *selectionStore
	workflow      *Workflow
	manager       *agenttask.Manager
	backend       *agentworkspace.Backend
	projectStores map[string]*projectlog.Store
}

// NewReader opens the authoritative state and workflow projection without
// creating overlay roots, provider processes, leases, or event records.
func NewReader(snapshot agentconfig.RuntimeSnapshot) (*Runtime, error) {
	if snapshot.Revision() == "" {
		return nil, errors.New("taskloop: immutable runtime snapshot is required")
	}
	cfg := snapshot.Config()
	if cfg.Device.StateRoot == "" {
		return nil, errors.New("taskloop: state_root is required")
	}
	state, err := agentstate.NewStore(filepath.Join(cfg.Device.StateRoot, "state.json"))
	if err != nil {
		return nil, err
	}
	selections := &selectionStore{state: state}
	workflow, err := NewWorkflow(snapshot, selections)
	if err != nil {
		return nil, err
	}
	return &Runtime{
		snapshot: snapshot, state: state, selections: selections, workflow: workflow,
	}, nil
}

type ProjectView struct {
	ProjectID         string
	Status            agenttask.ProjectStatus
	SelectedMilestone string
	StartedMilestone  string
	Works             []WorkView
	Blockers          []agenttask.Blocker
	StateRevision     agenttask.StateRevision
}

type WorkView struct {
	WorkUnitID      string
	State           agenttask.WorkState
	Overlay         string
	Integration     string
	DispatchOrdinal uint64
	Blocker         *agenttask.Blocker
}

type Preview struct {
	ProjectID string
	Selected  bool
	Milestone string
	NextWork  string
	Blockers  []agenttask.Blocker
}

func New(config Config) (*Runtime, error) {
	if config.Snapshot.Revision() == "" {
		return nil, errors.New("taskloop: immutable runtime snapshot is required")
	}
	cfg := config.Snapshot.Config()
	if cfg.Device.StateRoot == "" || cfg.Device.OverlayRoot == "" {
		return nil, errors.New("taskloop: state_root and overlay_root are required")
	}
	if config.Validator == nil {
		return nil, errors.New("taskloop: post-apply validator is required")
	}
	stateStore := config.StateStore
	if stateStore == nil {
		var err error
		stateStore, err = agentstate.NewStore(filepath.Join(cfg.Device.StateRoot, "state.json"))
		if err != nil {
			return nil, err
		}
	}
	normalizedCatalog, err := agentconfig.Normalize(config.Catalog)
	if err != nil {
		return nil, fmt.Errorf("taskloop: provider catalog: %w", err)
	}
	selections := &selectionStore{state: stateStore}
	workflow, err := NewWorkflow(config.Snapshot, selections)
	if err != nil {
		return nil, err
	}
	policySnapshot, err := selectionPolicySnapshot(config.Snapshot, normalizedCatalog)
	if err != nil {
		return nil, err
	}
	quotaObserver := config.QuotaObserver
	if quotaObserver == nil {
		quotaObserver = &catalogQuotaObserver{catalog: normalizedCatalog}
	}
	selector := &configSelector{
		snapshot:       config.Snapshot,
		policySnapshot: policySnapshot,
		catalog:        normalizedCatalog,
		state:          stateStore,
		evaluator:      agentpolicy.NewEvaluator(),
		now:            time.Now,
		quota:          quotaObserver,
	}
	resolver := agentworkspace.InputResolverFunc(func(
		ctx context.Context,
		request agenttask.IsolationRequest,
	) (agentworkspace.ResolvedInputs, error) {
		if err := ctx.Err(); err != nil {
			return agentworkspace.ResolvedInputs{}, err
		}
		registration, ok := config.Snapshot.Project(string(request.Project.ProjectID))
		if !ok {
			return agentworkspace.ResolvedInputs{}, errors.New("taskloop: isolation project is not registered")
		}
		root, err := canonicalDirectory(registration.Workspace)
		if err != nil {
			return agentworkspace.ResolvedInputs{}, err
		}
		if WorkspaceIdentity(root) != request.Project.WorkspaceID || request.Project.Intent == nil {
			return agentworkspace.ResolvedInputs{}, errors.New("taskloop: isolation workspace identity drift")
		}
		resolved, ok := normalizedCatalog.ResolveProfile(request.Target.ProfileID)
		if !ok {
			return agentworkspace.ResolvedInputs{}, errors.New("taskloop: isolation profile is not declared")
		}
		profile := guardProfile(resolved)
		if profile.Revision != request.Target.ProfileRevision {
			return agentworkspace.ResolvedInputs{}, errors.New("taskloop: isolation profile revision drift")
		}
		return agentworkspace.ResolvedInputs{
			Grant: agentguard.WorkspaceGrant{
				ProjectID:   string(request.Project.ProjectID),
				WorkspaceID: string(request.Project.WorkspaceID),
				Root:        root,
				Revision:    string(request.Project.Intent.GrantRevision),
			},
			Profile: profile,
		}, nil
	})
	backend, err := agentworkspace.NewBackend(agentworkspace.BackendConfig{
		LocalRoot: cfg.Device.OverlayRoot,
		Retention: cfg.Retention,
	}, resolver)
	if err != nil {
		return nil, err
	}
	provider := config.Provider
	if provider == nil {
		provider, err = NewProvider(normalizedCatalog, quotaObserver)
		if err != nil {
			return nil, err
		}
	}
	reviewExecutor := config.ReviewExecutor
	if reviewExecutor == nil {
		reviewExecutor, err = NewCatalogReviewExecutor(normalizedCatalog, backend)
		if err != nil {
			return nil, err
		}
	}
	reviewer, err := NewReviewer(config.Snapshot, backend, reviewExecutor)
	if err != nil {
		return nil, err
	}
	repairer, err := NewPiEvidenceRepairer(normalizedCatalog, backend)
	if err != nil {
		return nil, err
	}
	integration, err := NewIntegration(backend, stateStore, config.Validator)
	if err != nil {
		return nil, err
	}
	projectStores, sink, err := buildProjectEventSink(config.Snapshot, stateStore)
	if err != nil {
		return nil, err
	}
	eventSink := config.EventSink
	if eventSink == nil {
		eventSink = sink
	}
	ownerID := strings.TrimSpace(config.OwnerID)
	if ownerID == "" {
		ownerID = "iop-agent-standalone"
	}
	manager, err := agenttask.NewManager(
		agenttask.ManagerConfig{
			OwnerID:            ownerID,
			LeaseDuration:      30 * time.Second,
			MaxReworkAttempts:  3,
			MaxFailureAttempts: 10,
			StateWriteAttempts: 32,
		},
		nil,
		stateStore,
		workflow,
		selector,
		backend,
		provider,
		NewRecovery(config.Snapshot, retainedArtifactResolver{backend: backend}),
		NewEvidence(config.Snapshot, retainedArtifactResolver{backend: backend}, repairer),
		reviewer,
		integration,
		eventSink,
	)
	if err != nil {
		return nil, err
	}
	return &Runtime{
		snapshot:      config.Snapshot,
		catalog:       normalizedCatalog,
		state:         stateStore,
		selections:    selections,
		workflow:      workflow,
		manager:       manager,
		backend:       backend,
		projectStores: projectStores,
	}, nil
}

func (runtime *Runtime) Manager() agenttask.AgentTaskManager { return runtime.manager }
func (runtime *Runtime) StateStore() *agentstate.Store       { return runtime.state }
func (runtime *Runtime) ProjectStores() map[string]*projectlog.Store {
	out := make(map[string]*projectlog.Store, len(runtime.projectStores))
	for id, store := range runtime.projectStores {
		out[id] = store
	}
	return out
}

func (runtime *Runtime) StartProject(ctx context.Context, projectID string) (ProjectView, error) {
	if runtime.manager == nil {
		return ProjectView{}, errors.New("taskloop: runtime mutation owner is unavailable")
	}
	registration, selected, err := runtime.registeredSelection(ctx, projectID)
	if err != nil {
		return ProjectView{}, err
	}
	if selected == "" {
		return ProjectView{}, fmt.Errorf("project %s has no selected milestone", projectID)
	}
	snapshot, err := runtime.workflow.Snapshot(ctx, agenttask.ProjectID(projectID))
	if err != nil {
		return ProjectView{}, err
	}
	autoResume := registration.AutoResumeInterrupted
	err = runtime.manager.StartProject(ctx, agenttask.StartRequest{
		CommandID: agenttask.CommandID(digestStrings(
			"start", projectID, selected, string(snapshot.Revision), runtime.snapshot.Revision(),
		)),
		ProjectID:             agenttask.ProjectID(projectID),
		WorkspaceID:           snapshot.WorkspaceID,
		MilestoneID:           agenttask.MilestoneID(selected),
		WorkflowRevision:      snapshot.Revision,
		ConfigRevision:        agenttask.ConfigRevision(runtime.snapshot.Revision()),
		GrantRevision:         agenttask.GrantRevision(grantRevision(projectID, snapshot.WorkspaceID, runtime.snapshot.Revision())),
		AutoResumeInterrupted: &autoResume,
	})
	if err != nil {
		return ProjectView{}, err
	}
	return runtime.ProjectStatus(ctx, projectID)
}

func (runtime *Runtime) StopProject(ctx context.Context, projectID string) (ProjectView, error) {
	if runtime.manager == nil {
		return ProjectView{}, errors.New("taskloop: runtime mutation owner is unavailable")
	}
	if _, _, err := runtime.registeredSelection(ctx, projectID); err != nil {
		return ProjectView{}, err
	}
	if err := runtime.manager.StopProject(ctx, agenttask.ProjectID(projectID)); err != nil {
		return ProjectView{}, err
	}
	return runtime.ProjectStatus(ctx, projectID)
}

func (runtime *Runtime) ResumeProject(ctx context.Context, projectID string) (ProjectView, error) {
	if runtime.manager == nil {
		return ProjectView{}, errors.New("taskloop: runtime mutation owner is unavailable")
	}
	registration, selected, err := runtime.registeredSelection(ctx, projectID)
	if err != nil {
		return ProjectView{}, err
	}
	state, revision, err := runtime.state.Load(ctx)
	if err != nil {
		return ProjectView{}, err
	}
	project, ok := state.Projects[agenttask.ProjectID(projectID)]
	if ok && project.Intent != nil {
		if selected != string(project.Intent.MilestoneID) {
			return ProjectView{}, fmt.Errorf(
				"project %s selection changed from started milestone %s to %s; issue a new start instead of resume",
				projectID,
				project.Intent.MilestoneID,
				selected,
			)
		}
	}
	if selected == "" {
		return ProjectView{}, fmt.Errorf("project %s has no selected milestone", projectID)
	}
	snapshot, err := runtime.workflow.Snapshot(ctx, agenttask.ProjectID(projectID))
	if err != nil {
		return ProjectView{}, err
	}
	autoResume := registration.AutoResumeInterrupted
	err = runtime.manager.StartProject(ctx, agenttask.StartRequest{
		CommandID: agenttask.CommandID(digestStrings(
			"resume", projectID, selected, string(revision), string(snapshot.Revision),
		)),
		ProjectID:             agenttask.ProjectID(projectID),
		WorkspaceID:           snapshot.WorkspaceID,
		MilestoneID:           agenttask.MilestoneID(selected),
		WorkflowRevision:      snapshot.Revision,
		ConfigRevision:        agenttask.ConfigRevision(runtime.snapshot.Revision()),
		GrantRevision:         agenttask.GrantRevision(grantRevision(projectID, snapshot.WorkspaceID, runtime.snapshot.Revision())),
		AutoResumeInterrupted: &autoResume,
	})
	if err != nil {
		return ProjectView{}, err
	}
	return runtime.ProjectStatus(ctx, projectID)
}

func (runtime *Runtime) Reconcile(ctx context.Context) error {
	if runtime.manager == nil {
		return errors.New("taskloop: runtime mutation owner is unavailable")
	}
	if err := runtime.manager.Reconcile(ctx); err != nil {
		return err
	}
	state, _, err := runtime.state.Load(ctx)
	if err != nil {
		return err
	}
	return runtime.archiveTerminalProjectLogs(ctx, state)
}

func (runtime *Runtime) archiveTerminalProjectLogs(
	ctx context.Context,
	state agenttask.ManagerState,
) error {
	for projectID, project := range state.Projects {
		store, ok := runtime.projectStores[string(projectID)]
		if !ok {
			continue
		}
		for workUnitID, work := range project.Works {
			if !work.State.Terminal() {
				continue
			}
			scoped, err := store.ForWorkUnit(workUnitID)
			if err != nil {
				return err
			}
			if err := scoped.Archive(ctx); err != nil &&
				!errors.Is(err, projectlog.ErrNoTerminalRecord) {
				return fmt.Errorf(
					"taskloop: archive terminal project log %s/%s: %w",
					projectID,
					workUnitID,
					err,
				)
			}
		}
	}
	return nil
}

func (runtime *Runtime) SelectMilestone(
	ctx context.Context,
	projectID, milestone string,
) error {
	registration, ok := runtime.snapshot.Project(projectID)
	if !ok || !registration.Enabled {
		return fmt.Errorf("project %s is not registered", projectID)
	}
	root, err := canonicalDirectory(registration.Workspace)
	if err != nil {
		return err
	}
	if _, _, err := scanWorkflow(root, milestone); err != nil {
		return err
	}
	return runtime.selections.Save(ctx, projectID, milestone)
}

func (runtime *Runtime) Milestones(
	ctx context.Context,
	projectID string,
) ([]MilestoneView, error) {
	registration, selected, err := runtime.registeredSelection(ctx, projectID)
	if err != nil {
		return nil, err
	}
	root, err := canonicalDirectory(registration.Workspace)
	if err != nil {
		return nil, err
	}
	return scanMilestones(root, selected)
}

func (runtime *Runtime) WorkflowSnapshot(
	ctx context.Context,
	projectID string,
) (agenttask.ProjectWorkflowSnapshot, error) {
	return runtime.workflow.Snapshot(ctx, agenttask.ProjectID(projectID))
}

func (runtime *Runtime) Preview(ctx context.Context, projectID string) (Preview, error) {
	_, selected, err := runtime.registeredSelection(ctx, projectID)
	if err != nil {
		return Preview{}, err
	}
	result := Preview{ProjectID: projectID, Selected: selected != "", Milestone: selected}
	if selected == "" {
		return result, nil
	}
	snapshot, err := runtime.workflow.Snapshot(ctx, agenttask.ProjectID(projectID))
	if err != nil {
		result.Blockers = append(result.Blockers, agenttask.Blocker{
			Code: agenttask.BlockerWorkflowUnavailable, Message: err.Error(), Retryable: true,
		})
		return result, nil
	}
	completed := make(map[string]bool)
	for _, unit := range snapshot.Units {
		if unit.Completed {
			completed[string(unit.ID)] = true
			for _, alias := range unit.Aliases {
				completed[alias] = true
			}
		}
	}
	for _, unit := range snapshot.Units {
		if unit.Completed {
			continue
		}
		ready := true
		for _, predecessor := range unit.ExplicitPredecessors {
			if !completed[predecessor.Ref] {
				ready = false
				result.Blockers = append(result.Blockers, agenttask.Blocker{
					Code:    agenttask.BlockerDependencyMissing,
					Message: fmt.Sprintf("explicit predecessor %q is not complete", predecessor.Ref),
				})
			}
		}
		if ready {
			result.NextWork = string(unit.ID)
			break
		}
	}
	return result, nil
}

func (runtime *Runtime) ProjectStatus(ctx context.Context, projectID string) (ProjectView, error) {
	registration, selected, err := runtime.registeredSelection(ctx, projectID)
	if err != nil {
		return ProjectView{}, err
	}
	_ = registration
	view := ProjectView{
		ProjectID: projectID, Status: agenttask.ProjectStatusObserved,
		SelectedMilestone: selected,
	}
	if _, statErr := os.Stat(runtime.state.Path()); errors.Is(statErr, os.ErrNotExist) {
		return view, nil
	} else if statErr != nil {
		return ProjectView{}, statErr
	}
	state, revision, err := runtime.state.Load(ctx)
	if err != nil {
		return ProjectView{}, err
	}
	view.StateRevision = revision
	project, ok := state.Projects[agenttask.ProjectID(projectID)]
	if !ok {
		return view, nil
	}
	view.Status = project.Status
	if project.Intent != nil {
		view.StartedMilestone = string(project.Intent.MilestoneID)
	}
	if project.Blocker != nil {
		view.Blockers = append(view.Blockers, *project.Blocker)
	}
	workIDs := make([]agenttask.WorkUnitID, 0, len(project.Works))
	for workID := range project.Works {
		workIDs = append(workIDs, workID)
	}
	sort.Slice(workIDs, func(left, right int) bool { return workIDs[left] < workIDs[right] })
	for _, workID := range workIDs {
		work := project.Works[workID]
		workView := WorkView{
			WorkUnitID:      string(workID),
			State:           work.State,
			DispatchOrdinal: uint64(work.DispatchOrdinal),
		}
		if work.Isolation != nil {
			workView.Overlay = string(work.Isolation.Mode)
		}
		if work.Integration != nil {
			workView.Integration = string(work.Integration.Outcome)
		}
		if work.Blocker != nil {
			blocker := *work.Blocker
			workView.Blocker = &blocker
			view.Blockers = append(view.Blockers, blocker)
		}
		view.Works = append(view.Works, workView)
	}
	return view, nil
}

func (runtime *Runtime) registeredSelection(
	ctx context.Context,
	projectID string,
) (agentconfig.ProjectRegistration, string, error) {
	registration, ok := runtime.snapshot.Project(projectID)
	if !ok || !registration.Enabled {
		return agentconfig.ProjectRegistration{}, "", fmt.Errorf("project %s is not registered", projectID)
	}
	selected, err := runtime.selections.SelectedMilestone(ctx, projectID)
	if err != nil {
		return agentconfig.ProjectRegistration{}, "", err
	}
	if selected == "" {
		selected = registration.SelectedMilestone
	}
	return registration, selected, nil
}

type configSelector struct {
	snapshot       agentconfig.RuntimeSnapshot
	policySnapshot agentconfig.RuntimeSnapshot
	catalog        agentconfig.Catalog
	state          *agentstate.Store
	evaluator      *agentpolicy.Evaluator
	now            func() time.Time
	quota          QuotaObserver
}

func (selector *configSelector) Select(
	ctx context.Context,
	request agenttask.SelectionRequest,
) (agenttask.ExecutionTarget, error) {
	if _, ok := selector.snapshot.Project(string(request.Project.ProjectID)); !ok {
		return agenttask.ExecutionTarget{}, errors.New("taskloop: selection project is not registered")
	}
	if selector.evaluator == nil || selector.state == nil ||
		selector.now == nil || selector.quota == nil {
		return agenttask.ExecutionTarget{}, errors.New("taskloop: policy selector is not fully configured")
	}
	selectionContext := selector.selectionContext(request)
	key := routeDecisionKey(request)
	persisted, _, found, err := selector.state.LoadIntegrationRecord(ctx, key)
	if err != nil {
		return agenttask.ExecutionTarget{}, err
	}
	var decision agentpolicy.RouteDecision
	if found {
		decision, err = selector.evaluator.ResumePolicy(
			ctx,
			selector.policySnapshot,
			selectionContext,
			persisted,
		)
	} else {
		decision, err = selector.evaluator.SelectPolicy(
			ctx,
			selector.policySnapshot,
			selectionContext,
		)
		if err == nil {
			decision, err = selector.persistRouteDecision(
				ctx,
				key,
				selectionContext,
				decision,
			)
		}
	}
	if err != nil {
		return agenttask.ExecutionTarget{}, fmt.Errorf("taskloop: select ordered provider policy: %w", err)
	}
	return selector.executionTarget(decision)
}

func (selector *configSelector) executionTarget(
	decision agentpolicy.RouteDecision,
) (agenttask.ExecutionTarget, error) {
	resolved, ok := selector.catalog.ResolveProfile(decision.ProfileID)
	if !ok {
		return agenttask.ExecutionTarget{}, fmt.Errorf(
			"taskloop: selected profile %q is not declared",
			decision.ProfileID,
		)
	}
	if decision.ProviderID != resolved.Provider.ID ||
		decision.ModelID != resolved.Model.ID ||
		decision.ConfigRevision != selector.policySnapshot.Revision() {
		return agenttask.ExecutionTarget{}, errors.New(
			"taskloop: durable route identity disagrees with the catalog",
		)
	}
	capacity := resolved.Profile.MaxConcurrency
	if capacity <= 0 {
		capacity = 1
	}
	return agenttask.ExecutionTarget{
		ProviderID:      resolved.Provider.ID,
		ModelID:         resolved.Model.ID,
		ProfileID:       resolved.Profile.ID,
		ProfileRevision: guardProfile(resolved).Revision,
		ConfigRevision:  agenttask.ConfigRevision(selector.snapshot.Revision()),
		Capacity:        capacity,
	}, nil
}

func (selector *configSelector) persistRouteDecision(
	ctx context.Context,
	key string,
	selectionContext agentpolicy.SelectionContext,
	decision agentpolicy.RouteDecision,
) (agentpolicy.RouteDecision, error) {
	payload, err := agentpolicy.EncodeDecision(decision)
	if err != nil {
		return agentpolicy.RouteDecision{}, err
	}
	for attempt := 0; attempt < 32; attempt++ {
		current, revision, found, err := selector.state.LoadIntegrationRecord(ctx, key)
		if err != nil {
			return agentpolicy.RouteDecision{}, err
		}
		if found {
			return selector.evaluator.ResumePolicy(
				ctx,
				selector.policySnapshot,
				selectionContext,
				current,
			)
		}
		if _, err := selector.state.CompareAndSwapIntegrationRecord(
			ctx,
			key,
			revision,
			payload,
		); errors.Is(err, agenttask.ErrRevisionConflict) {
			continue
		} else if err != nil {
			return agentpolicy.RouteDecision{}, err
		}
		return decision, nil
	}
	return agentpolicy.RouteDecision{}, errors.New(
		"taskloop: route decision CAS retries exhausted",
	)
}

func (selector *configSelector) selectionContext(
	request agenttask.SelectionRequest,
) agentpolicy.SelectionContext {
	lane, grade := routeLaneAndGrade(request.Work.Unit.Metadata["plan_path"])
	if explicit := strings.TrimSpace(request.Work.Unit.Metadata["lane"]); explicit != "" {
		lane = explicit
	}
	if explicit := strings.TrimSpace(request.Work.Unit.Metadata["grade"]); explicit != "" {
		if parsed, err := strconv.Atoi(explicit); err == nil {
			grade = parsed
		}
	}
	remainingTokens, _ := strconv.ParseInt(
		strings.TrimSpace(request.Work.Unit.Metadata["remaining_tokens"]),
		10,
		64,
	)
	var capabilities []string
	for _, capability := range strings.Split(
		request.Work.Unit.Metadata["required_capabilities"],
		",",
	) {
		if capability = strings.TrimSpace(capability); capability != "" {
			capabilities = append(capabilities, capability)
		}
	}
	return agentpolicy.SelectionContext{
		ProjectID:       string(request.Project.ProjectID),
		Now:             selector.now().UTC(),
		Stage:           "worker",
		Grade:           grade,
		Agent:           strings.TrimSpace(request.Work.Unit.Metadata["agent"]),
		Lane:            lane,
		QuotaState:      strings.TrimSpace(request.Work.Unit.Metadata["quota_state"]),
		RemainingTokens: remainingTokens,
		Capabilities:    capabilities,
	}
}

func routeLaneAndGrade(planPath string) (string, int) {
	name := strings.TrimSuffix(filepath.Base(planPath), filepath.Ext(planPath))
	if !strings.HasPrefix(name, "PLAN-") {
		return "", 0
	}
	route := strings.TrimPrefix(name, "PLAN-")
	gradeIndex := strings.LastIndex(route, "-G")
	if gradeIndex < 0 {
		return "", 0
	}
	grade, err := strconv.Atoi(route[gradeIndex+2:])
	if err != nil {
		return route[:gradeIndex], 0
	}
	return route[:gradeIndex], grade
}

func routeDecisionKey(request agenttask.SelectionRequest) string {
	return "taskloop-route/" + strings.TrimPrefix(digestStrings(
		"route",
		string(request.Project.ProjectID),
		string(request.Project.WorkspaceID),
		string(request.Work.Unit.ID),
		string(request.Work.AttemptID),
	), "sha256:")
}

func (selector *configSelector) ContinuationPolicy(
	ctx context.Context,
	request agenttask.FailureContinuationPolicyRequest,
) (agenttask.FailureContinuationPolicy, error) {
	registration, ok := selector.snapshot.Project(string(request.Project.ProjectID))
	if !ok {
		return agenttask.FailureContinuationPolicy{}, errors.New(
			"taskloop: continuation project is not registered",
		)
	}
	if selector.evaluator == nil || selector.now == nil || selector.quota == nil {
		return agenttask.FailureContinuationPolicy{}, errors.New(
			"taskloop: continuation policy selector is not fully configured",
		)
	}

	selectionContext := selector.selectionContext(agenttask.SelectionRequest{
		Project: request.Project,
		Work:    request.Work,
	})
	selectionContext.FailureCode = string(request.Observation.Failure.Code)
	selectionContext.QuotaState = string(request.Observation.Quota.State)
	decision, err := selector.evaluator.SelectPolicy(
		ctx,
		selector.policySnapshot,
		selectionContext,
	)
	if err != nil {
		return agenttask.FailureContinuationPolicy{}, fmt.Errorf(
			"taskloop: evaluate continuation policy: %w",
			err,
		)
	}

	var selectedRule *agentconfig.SelectionRule
	for index := range registration.Selection.Rules {
		rule := &registration.Selection.Rules[index]
		if rule.ID == decision.SelectedRuleID {
			selectedRule = rule
			break
		}
	}
	if selectedRule == nil ||
		!containsFailureCode(selectedRule.Match.FailureCodes, request.Observation.Failure.Code) {
		return agenttask.FailureContinuationPolicy{}, errors.New(
			"taskloop: continuation requires an explicitly matching failure rule",
		)
	}

	target, err := selector.executionTarget(decision)
	if err != nil {
		return agenttask.FailureContinuationPolicy{}, err
	}
	quota, err := selector.quota.ObserveQuota(ctx, target, selector.now().UTC())
	if err != nil {
		return agenttask.FailureContinuationPolicy{}, fmt.Errorf(
			"taskloop: observe continuation quota: %w",
			err,
		)
	}
	if _, err := json.Marshal(quota); err != nil {
		return agenttask.FailureContinuationPolicy{}, fmt.Errorf(
			"taskloop: invalid continuation quota evidence: %w",
			err,
		)
	}

	code := request.Observation.Failure.Code
	policy := agentpolicy.FailurePolicy{}
	if sameExecutionTarget(target, request.CurrentTarget) {
		policy.RetryableCodes = []agentruntime.FailureCode{code}
	} else {
		policy.FailoverCodes = []agentruntime.FailureCode{code}
	}
	return agenttask.FailureContinuationPolicy{
		Policy: policy,
		Candidates: []agenttask.FailureContinuationCandidate{{
			Target:   target,
			Eligible: !sameExecutionTarget(target, request.CurrentTarget),
			Quota:    quota,
		}},
	}, nil
}

func (selector *configSelector) executionTargetRef(
	reference agentconfig.TargetRef,
) (agenttask.ExecutionTarget, error) {
	profileID := reference.Profile
	if profileID == "" {
		return agenttask.ExecutionTarget{}, errors.New(
			"taskloop: continuation target profile is missing",
		)
	}
	resolved, ok := selector.catalog.ResolveProfile(profileID)
	if !ok {
		return agenttask.ExecutionTarget{}, fmt.Errorf(
			"taskloop: continuation profile %q is not declared",
			profileID,
		)
	}
	if reference.Provider != "" && reference.Provider != resolved.Provider.ID ||
		reference.Model != "" && reference.Model != resolved.Model.ID {
		return agenttask.ExecutionTarget{}, errors.New(
			"taskloop: continuation target identity disagrees with the catalog",
		)
	}
	capacity := resolved.Profile.MaxConcurrency
	if capacity <= 0 {
		capacity = 1
	}
	return agenttask.ExecutionTarget{
		ProviderID:      resolved.Provider.ID,
		ModelID:         resolved.Model.ID,
		ProfileID:       resolved.Profile.ID,
		ProfileRevision: guardProfile(resolved).Revision,
		ConfigRevision:  agenttask.ConfigRevision(selector.snapshot.Revision()),
		Capacity:        capacity,
	}, nil
}

func sameExecutionTarget(left, right agenttask.ExecutionTarget) bool {
	return left.ProviderID == right.ProviderID &&
		left.ModelID == right.ModelID &&
		left.ProfileID == right.ProfileID &&
		left.ProfileRevision == right.ProfileRevision
}

func containsFailureCode(
	codes []string,
	want agentruntime.FailureCode,
) bool {
	for _, code := range codes {
		if agentruntime.FailureCode(strings.TrimSpace(code)) == want {
			return true
		}
	}
	return false
}

type catalogQuotaObserver struct {
	catalog agentconfig.Catalog
}

func (observer *catalogQuotaObserver) ObserveQuota(
	ctx context.Context,
	target agenttask.ExecutionTarget,
	observedAt time.Time,
) (agentpolicy.QuotaObservation, error) {
	normalize := func(usage *clistatus.UsageStatus, checkErr error) agentpolicy.QuotaObservation {
		snapshot := clistatus.NormalizeQuotaSnapshot(
			target.ProviderID,
			target.ProfileID,
			[]string{"overall"},
			observedAt,
			usage,
			checkErr,
		)
		return agentpolicy.NormalizeQuotaObservation(snapshot, observedAt, time.Minute)
	}
	resolved, ok := observer.catalog.ResolveProfile(target.ProfileID)
	if !ok || validateCatalogTarget(resolved, target) != nil {
		return normalize(nil, errors.New("catalog target identity mismatch")), nil
	}
	discoverer, err := catalog.NewDiscoverer(observer.catalog, nil)
	if err != nil {
		return normalize(nil, err), nil
	}
	readiness, err := discoverer.DiscoverProfile(ctx, target.ProfileID)
	if err != nil {
		return normalize(nil, err), nil
	}
	provider, err := catalog.NewProfileProvider(
		observer.catalog,
		target.ProfileID,
		readiness,
		zap.NewNop(),
	)
	if err != nil {
		return normalize(nil, err), nil
	}
	response, err := provider.HandleCommand(ctx, agentruntime.CommandRequest{
		RequestID: "taskloop-quota-" + strings.TrimPrefix(
			digestStrings("quota", target.ProfileID, observedAt.UTC().Format(time.RFC3339Nano)),
			"sha256:",
		),
		Type:    agentruntime.CommandTypeUsageStatus,
		Adapter: "cli",
		Target:  target.ProfileID,
	})
	if err != nil || response.UsageStatus == nil {
		return normalize(nil, err), nil
	}
	usage := &clistatus.UsageStatus{
		DailyLimit:      response.UsageStatus.DailyLimit,
		DailyResetTime:  response.UsageStatus.DailyResetTime,
		WeeklyLimit:     response.UsageStatus.WeeklyLimit,
		WeeklyResetTime: response.UsageStatus.WeeklyResetTime,
		Metadata:        response.UsageStatus.Metadata,
	}
	return normalize(usage, nil), nil
}

func selectionPolicySnapshot(
	snapshot agentconfig.RuntimeSnapshot,
	providerCatalog agentconfig.Catalog,
) (agentconfig.RuntimeSnapshot, error) {
	cfg := snapshot.Config()
	global := agentconfig.RepoGlobalRuntimeConfig{
		Version:   agentconfig.RuntimeConfigSchemaVersion,
		Catalog:   providerCatalog,
		Defaults:  cfg.Defaults,
		Selection: cfg.Selection,
		Isolation: cfg.Isolation,
		Retention: cfg.Retention,
	}
	local := agentconfig.UserLocalRuntimeConfig{
		Version:  agentconfig.RuntimeConfigSchemaVersion,
		Device:   cfg.Device,
		Clients:  cfg.Clients,
		Projects: make(map[string]agentconfig.ProjectRegistrationOverlay, len(cfg.Projects)),
	}
	for projectID, project := range cfg.Projects {
		enabled := project.Enabled
		defaultProfile := project.Defaults.DefaultProfile
		autoResume := project.Defaults.AutoResumeInterrupted
		timezone := project.Selection.Timezone
		defaultTarget := project.Selection.Default
		rules := project.Selection.Rules
		defaultMode := project.Isolation.DefaultMode
		fallbackModes := project.Isolation.FallbackModes
		completedDays := project.Retention.CompletedDays
		blockedDays := project.Retention.BlockedDays
		maxLogRecords := project.Retention.MaxProjectLogRecords
		local.Projects[projectID] = agentconfig.ProjectRegistrationOverlay{
			Workspace:         project.Workspace,
			Enabled:           &enabled,
			SelectedMilestone: project.SelectedMilestone,
			Override: agentconfig.RuntimeConfigOverride{
				Defaults: agentconfig.RuntimeDefaultsOverride{
					DefaultProfile:        &defaultProfile,
					AutoResumeInterrupted: &autoResume,
					ProfileAliases:        project.Defaults.ProfileAliases,
				},
				Selection: agentconfig.SelectionPolicyOverride{
					Timezone: &timezone,
					Default:  &defaultTarget,
					Rules:    &rules,
				},
				Isolation: agentconfig.IsolationPolicyOverride{
					DefaultMode:   &defaultMode,
					FallbackModes: &fallbackModes,
				},
				Retention: agentconfig.RetentionPolicyOverride{
					CompletedDays:        &completedDays,
					BlockedDays:          &blockedDays,
					MaxProjectLogRecords: &maxLogRecords,
				},
			},
		}
	}
	globalData, err := yaml.Marshal(global)
	if err != nil {
		return agentconfig.RuntimeSnapshot{}, fmt.Errorf(
			"taskloop: encode policy runtime config: %w",
			err,
		)
	}
	localData, err := yaml.Marshal(local)
	if err != nil {
		return agentconfig.RuntimeSnapshot{}, fmt.Errorf(
			"taskloop: encode policy device config: %w",
			err,
		)
	}
	policySnapshot, err := agentconfig.LoadRuntimeConfigBytes(globalData, localData)
	if err != nil {
		return agentconfig.RuntimeSnapshot{}, fmt.Errorf(
			"taskloop: compose catalog-backed policy snapshot: %w",
			err,
		)
	}
	return policySnapshot, nil
}

func guardProfile(resolved agentconfig.ResolvedProfile) agentguard.ProviderProfile {
	capabilities := make(map[string]bool)
	for _, capability := range resolved.Profile.Capabilities {
		capabilities[capability] = true
	}
	return agentguard.ProviderProfile{
		ProviderID:              resolved.Provider.ID,
		ModelID:                 resolved.Model.ID,
		ProfileID:               resolved.Profile.ID,
		Revision:                digestStrings("profile", resolved.Provider.ID, resolved.Model.ID, resolved.Profile.ID, strings.Join(resolved.Profile.Capabilities, ",")),
		Unattended:              capabilities["unattended"],
		ApprovalBypass:          capabilities["approval_bypass"],
		WritableRootConfinement: capabilities["writable_root_confinement"],
	}
}

type selectionRecord struct {
	SchemaVersion uint32 `json:"schema_version"`
	ProjectID     string `json:"project_id"`
	MilestoneID   string `json:"milestone_id"`
}

type selectionStore struct {
	state *agentstate.Store
}

func (store *selectionStore) SelectedMilestone(
	ctx context.Context,
	projectID string,
) (string, error) {
	if _, err := os.Stat(store.state.Path()); errors.Is(err, os.ErrNotExist) {
		return "", nil
	} else if err != nil {
		return "", err
	}
	payload, _, found, err := store.state.LoadIntegrationRecord(ctx, selectionKey(projectID))
	if err != nil || !found {
		return "", err
	}
	var record selectionRecord
	if err := decodeStrictJSON(payload, &record); err != nil {
		return "", fmt.Errorf("taskloop: decode selected milestone: %w", err)
	}
	if record.SchemaVersion != 1 || record.ProjectID != projectID ||
		strings.TrimSpace(record.MilestoneID) == "" {
		return "", errors.New("taskloop: selected milestone record is corrupt")
	}
	return record.MilestoneID, nil
}

func (store *selectionStore) Save(
	ctx context.Context,
	projectID, milestone string,
) error {
	record := selectionRecord{SchemaVersion: 1, ProjectID: projectID, MilestoneID: milestone}
	payload, err := json.Marshal(record)
	if err != nil {
		return err
	}
	for attempt := 0; attempt < 32; attempt++ {
		_, revision, found, err := store.state.LoadIntegrationRecord(ctx, selectionKey(projectID))
		if err != nil {
			return err
		}
		if !found {
			revision = ""
		}
		if _, err := store.state.CompareAndSwapIntegrationRecord(
			ctx, selectionKey(projectID), revision, payload,
		); errors.Is(err, agenttask.ErrRevisionConflict) {
			continue
		} else {
			return err
		}
	}
	return errors.New("taskloop: selected milestone CAS retries exhausted")
}

func selectionKey(projectID string) string {
	return "taskloop-selection/" + strings.TrimPrefix(digestStrings("project", projectID), "sha256:")
}

func grantRevision(projectID string, workspaceID agenttask.WorkspaceID, configRevision string) string {
	return digestStrings("grant", projectID, string(workspaceID), configRevision)
}

type projectEventSink struct {
	sinks map[agenttask.ProjectID]agenttask.EventSink
}

func (sink *projectEventSink) Emit(ctx context.Context, event agenttask.Event) error {
	projectSink, ok := sink.sinks[event.ProjectID]
	if !ok {
		return fmt.Errorf("taskloop: project log sink for %q is unavailable", event.ProjectID)
	}
	return projectSink.Emit(ctx, event)
}

func buildProjectEventSink(
	snapshot agentconfig.RuntimeSnapshot,
	state *agentstate.Store,
) (map[string]*projectlog.Store, agenttask.EventSink, error) {
	cfg := snapshot.Config()
	logRoot := cfg.Device.LogRoot
	if logRoot == "" {
		logRoot = filepath.Join(cfg.Device.StateRoot, "logs")
	}
	resolver, err := projectlog.NewStateStoreEvidenceResolver(state)
	if err != nil {
		return nil, nil, err
	}
	stores := make(map[string]*projectlog.Store, len(cfg.Projects))
	sinks := make(map[agenttask.ProjectID]agenttask.EventSink, len(cfg.Projects))
	for id, registration := range cfg.Projects {
		root := registration.Workspace
		if canonical, err := canonicalDirectory(root); err == nil {
			root = canonical
		}
		store, err := projectlog.NewStore(
			state,
			logRoot,
			agenttask.ProjectID(id),
			WorkspaceIdentity(root),
		)
		if err != nil {
			return nil, nil, err
		}
		eventSink, err := projectlog.NewSink(store, resolver)
		if err != nil {
			return nil, nil, err
		}
		stores[id] = store
		sinks[agenttask.ProjectID(id)] = eventSink
	}
	return stores, &projectEventSink{sinks: sinks}, nil
}

// Component drives reconciliation only during the sustained daemon lifecycle.
type Component struct {
	runtime  *Runtime
	interval time.Duration
	cancel   context.CancelFunc
	done     chan struct{}
	errMu    sync.Mutex
	lastErr  error
}

func NewComponent(runtime *Runtime, interval time.Duration) (*Component, error) {
	if runtime == nil {
		return nil, errors.New("taskloop: runtime component requires a runtime")
	}
	if interval <= 0 {
		interval = time.Second
	}
	return &Component{runtime: runtime, interval: interval}, nil
}

func (component *Component) Name() string { return "task-runtime" }

func (component *Component) Start(ctx context.Context) error {
	if component.cancel != nil {
		return errors.New("taskloop: runtime component already started")
	}
	runCtx, cancel := context.WithCancel(ctx)
	component.cancel = cancel
	component.done = make(chan struct{})
	go component.run(runCtx)
	return nil
}

func (component *Component) run(ctx context.Context) {
	defer close(component.done)
	ticker := time.NewTicker(component.interval)
	defer ticker.Stop()
	for {
		if err := component.runtime.Reconcile(ctx); err != nil && ctx.Err() == nil {
			component.errMu.Lock()
			component.lastErr = err
			component.errMu.Unlock()
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func (component *Component) Stop(ctx context.Context) error {
	if component.cancel == nil {
		return nil
	}
	component.cancel()
	select {
	case <-component.done:
	case <-ctx.Done():
		return ctx.Err()
	}
	component.cancel = nil
	// Reconciliation failures are committed as project blockers by the shared
	// manager. They are not lifecycle shutdown failures and must not make a
	// clean daemon stop fail retroactively.
	return nil
}

func decodeStrictJSON(payload []byte, destination any) error {
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(destination); err != nil {
		return err
	}
	var trailing any
	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
		if err == nil {
			return errors.New("multiple JSON values are not allowed")
		}
		return err
	}
	return nil
}