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

package taskloop

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"iop/packages/go/agentconfig"
	"iop/packages/go/agentpolicy"
	"iop/packages/go/agentruntime"
	"iop/packages/go/agenttask"
)

const (
	maxProviderDiagnosticBytes        = 32 << 10
	nativeSessionLocatorSchemaVersion = 1
)

// Provider converts one catalog profile into the side-effect-free launch plan
// required by agenttask.Manager. The executable confinement proof remains the
// sole process and stdio owner.
type Provider struct {
	catalog agentconfig.Catalog
	quota   QuotaObserver
	now     func() time.Time
}

var _ agenttask.ProviderInvoker = (*Provider)(nil)

func NewProvider(
	catalog agentconfig.Catalog,
	quotaObservers ...QuotaObserver,
) (*Provider, error) {
	normalized, err := agentconfig.Normalize(catalog)
	if err != nil {
		return nil, fmt.Errorf("taskloop: normalize provider catalog: %w", err)
	}
	var quota QuotaObserver
	if len(quotaObservers) > 1 {
		return nil, errors.New("taskloop: provider accepts at most one quota observer")
	}
	if len(quotaObservers) == 1 {
		quota = quotaObservers[0]
	}
	return &Provider{catalog: normalized, quota: quota, now: time.Now}, nil
}

func (provider *Provider) Prepare(
	ctx context.Context,
	request agenttask.DispatchRequest,
) (agenttask.ProviderLaunch, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if request.Permit == nil || request.Confinement == nil {
		return nil, errors.New("taskloop: provider launch requires an admitted confinement")
	}
	resolved, ok := provider.catalog.ResolveProfile(request.Target.ProfileID)
	if !ok {
		return nil, fmt.Errorf(
			"taskloop: selected provider profile %q is not in the catalog",
			request.Target.ProfileID,
		)
	}
	if resolved.Provider.ID != request.Target.ProviderID ||
		resolved.Model.ID != request.Target.ModelID {
		return nil, errors.New("taskloop: selected provider identity does not match the catalog")
	}
	plan := request.Work.Unit.Metadata["plan_path"]
	review := request.Work.Unit.Metadata["review_path"]
	if plan == "" || review == "" {
		return nil, errors.New("taskloop: work unit is missing its active artifact locators")
	}
	prompt := workerPrompt(plan, review)
	binding := request.Confinement.Binding()
	if request.Confinement.Revision() != binding.Revision ||
		binding.ProfileRevision != request.Target.ProfileRevision {
		return nil, errors.New("taskloop: admitted confinement identity mismatch")
	}
	if err := request.Confinement.Validate(binding); err != nil {
		return nil, fmt.Errorf("taskloop: validate admitted confinement: %w", err)
	}
	command, session, err := prepareFreshCatalogCommand(
		resolved,
		request.Target,
		binding,
		request.Workspace.WorkingDir,
		prompt,
		request.Project,
		request.Work,
		"worker",
	)
	if err != nil {
		return nil, err
	}
	return &providerLaunch{
		command:   command,
		project:   request.Project,
		work:      request.Work,
		artifact:  artifactIdentity(request.Work),
		profileID: resolved.Profile.ID,
		session:   session,
		target:    request.Target,
		quota:     provider.quota,
		now:       provider.now,
	}, nil
}

func workerPrompt(plan, review string) string {
	return fmt.Sprintf(
		"This is a direct implementing-worker assignment owned by the task loop. "+
			"Read %s and execute it directly; do not invoke a task-loop dispatcher, "+
			"plan workflow, or code-review workflow. Keep artifact content in English. "+
			"Fill only the implementation-owned sections in %s and stop ready for "+
			"official review. Do not append a review verdict, modify review-only fields, "+
			"write completion metadata, update a roadmap, or archive, move, rename, or "+
			"delete either active artifact.",
		plan,
		review,
	)
}

type providerLaunch struct {
	command   agenttask.ConfinementCommand
	project   agenttask.ProjectRecord
	work      agenttask.WorkRecord
	artifact  agenttask.ArtifactID
	profileID string
	session   *agenttask.LocatorRecord
	target    agenttask.ExecutionTarget
	quota     QuotaObserver
	now       func() time.Time
}

func (launch *providerLaunch) Command() agenttask.ConfinementCommand {
	return agenttask.ConfinementCommand{
		Name: launch.command.Name,
		Args: append([]string(nil), launch.command.Args...),
		Env:  append([]string(nil), launch.command.Env...),
	}
}

func (launch *providerLaunch) BindStarted(
	started agenttask.StartedConfinement,
) (agenttask.ProviderInvocation, error) {
	if started == nil || started.Child() == nil || started.Child().Process == nil ||
		started.Stdin() == nil || started.Stdout() == nil || started.Stderr() == nil {
		return nil, errors.New("taskloop: confined provider child is incomplete")
	}
	opaque, err := processLocator(started.Child().Process.Pid)
	if err != nil {
		return nil, err
	}
	locator := agenttask.LocatorRecord{
		Kind:        agenttask.LocatorProcess,
		Opaque:      opaque,
		Revision:    digestStrings("process-locator", opaque),
		ProjectID:   launch.project.ProjectID,
		WorkspaceID: launch.project.WorkspaceID,
		WorkUnitID:  launch.work.Unit.ID,
		AttemptID:   launch.work.AttemptID,
	}
	locators := []agenttask.LocatorRecord{locator}
	if launch.session != nil {
		locators = append(locators, *launch.session)
	}
	invocation := &providerInvocation{
		started:  started,
		project:  launch.project,
		work:     launch.work,
		artifact: launch.artifact,
		locators: locators,
		target:   launch.target,
		quota:    launch.quota,
		now:      launch.now,
		done:     make(chan invocationResult, 1),
	}
	invocation.begin()
	return invocation, nil
}

type invocationResult struct {
	err error
}

type providerInvocation struct {
	started   agenttask.StartedConfinement
	project   agenttask.ProjectRecord
	work      agenttask.WorkRecord
	artifact  agenttask.ArtifactID
	locators  []agenttask.LocatorRecord
	target    agenttask.ExecutionTarget
	quota     QuotaObserver
	now       func() time.Time
	stdout    boundedBuffer
	stderr    boundedBuffer
	done      chan invocationResult
	once      sync.Once
	failureMu sync.Mutex
	failure   agentpolicy.AttemptObservation
}

var _ agenttask.ProviderInvocation = (*providerInvocation)(nil)
var _ agenttask.FailureObservedInvocation = (*providerInvocation)(nil)

func (invocation *providerInvocation) begin() {
	invocation.once.Do(func() {
		_ = invocation.started.Stdin().Close()
		var drains sync.WaitGroup
		drains.Add(2)
		go func() {
			defer drains.Done()
			_, _ = io.Copy(&invocation.stdout, invocation.started.Stdout())
			_ = invocation.started.Stdout().Close()
		}()
		go func() {
			defer drains.Done()
			_, _ = io.Copy(&invocation.stderr, invocation.started.Stderr())
			_ = invocation.started.Stderr().Close()
		}()
		go func() {
			waitErr := invocation.started.Child().Wait()
			drains.Wait()
			invocation.done <- invocationResult{err: waitErr}
			close(invocation.done)
		}()
	})
}

func (invocation *providerInvocation) Locators() []agenttask.LocatorRecord {
	return append([]agenttask.LocatorRecord(nil), invocation.locators...)
}

func (invocation *providerInvocation) Wait(
	ctx context.Context,
) (agenttask.Submission, error) {
	select {
	case <-ctx.Done():
		_ = invocation.Cancel(context.WithoutCancel(ctx))
		invocation.recordFailure(
			context.WithoutCancel(ctx),
			&agentruntime.Failure{
				Code:      agentruntime.FailureCodeCancelled,
				Retryable: false,
			},
			false,
		)
		return agenttask.Submission{}, ctx.Err()
	case result := <-invocation.done:
		if result.err != nil {
			if ctx.Err() != nil {
				invocation.recordFailure(
					context.WithoutCancel(ctx),
					&agentruntime.Failure{
						Code:      agentruntime.FailureCodeCancelled,
						Retryable: false,
					},
					false,
				)
				return agenttask.Submission{}, ctx.Err()
			}
			invocation.recordFailure(
				ctx,
				&agentruntime.Failure{
					Code:      agentruntime.FailureCodeProcessExit,
					Retryable: true,
				},
				true,
			)
			return agenttask.Submission{}, errors.New("taskloop: provider exited unsuccessfully")
		}
		return agenttask.Submission{
			ProjectID:  invocation.project.ProjectID,
			WorkUnitID: invocation.work.Unit.ID,
			AttemptID:  invocation.work.AttemptID,
			ArtifactID: invocation.artifact,
			Ready:      true,
			Metadata: map[string]string{
				"provider_output": "bounded-and-discarded",
			},
			Locators: invocation.Locators(),
		}, nil
	}
}

func (invocation *providerInvocation) FailureObservation() agentpolicy.AttemptObservation {
	invocation.failureMu.Lock()
	defer invocation.failureMu.Unlock()
	return invocation.failure.Clone()
}

func (invocation *providerInvocation) recordFailure(
	ctx context.Context,
	failure *agentruntime.Failure,
	observeQuota bool,
) {
	now := time.Now
	if invocation.now != nil {
		now = invocation.now
	}
	observedAt := now().UTC()
	quota := agentpolicy.CorruptQuotaObservation()
	if observeQuota && invocation.quota != nil {
		if observed, err := invocation.quota.ObserveQuota(
			ctx,
			invocation.target,
			observedAt,
		); err == nil {
			quota = observed
		}
	}
	invocation.failureMu.Lock()
	invocation.failure = agentpolicy.SanitizeAttemptObservation(
		agentpolicy.AttemptObservation{
			ObservedAt: observedAt,
			Quota:      quota,
			Failure:    agentpolicy.NormalizeFailureObservation(failure),
		},
		observedAt,
	)
	invocation.failureMu.Unlock()
}

func (invocation *providerInvocation) Cancel(ctx context.Context) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	child := invocation.started.Child()
	if child == nil || child.Process == nil {
		return errors.New("taskloop: provider process identity is unavailable")
	}
	if err := child.Process.Signal(syscall.SIGTERM); err != nil &&
		!errors.Is(err, os.ErrProcessDone) {
		if killErr := child.Process.Kill(); killErr != nil &&
			!errors.Is(killErr, os.ErrProcessDone) {
			return errors.Join(err, killErr)
		}
	}
	return nil
}

type boundedBuffer struct {
	mu        sync.Mutex
	buffer    bytes.Buffer
	truncated bool
}

func (buffer *boundedBuffer) Write(data []byte) (int, error) {
	buffer.mu.Lock()
	defer buffer.mu.Unlock()
	remaining := maxProviderDiagnosticBytes - buffer.buffer.Len()
	if remaining > 0 {
		write := len(data)
		if write > remaining {
			write = remaining
		}
		_, _ = buffer.buffer.Write(data[:write])
	}
	if len(data) > remaining {
		buffer.truncated = true
	}
	return len(data), nil
}

type processReference struct {
	PID        int    `json:"pid"`
	StartToken string `json:"start_token"`
}

func processLocator(pid int) (string, error) {
	if pid <= 0 {
		return "", errors.New("taskloop: provider PID is invalid")
	}
	token, err := processStartToken(pid)
	if err != nil {
		return "", fmt.Errorf("taskloop: capture provider process identity: %w", err)
	}
	encoded, err := json.Marshal(processReference{PID: pid, StartToken: token})
	if err != nil {
		return "", err
	}
	return string(encoded), nil
}

func processStartToken(pid int) (string, error) {
	if stat, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/stat"); err == nil {
		text := string(stat)
		if closeIndex := strings.LastIndex(text, ")"); closeIndex >= 0 {
			fields := strings.Fields(text[closeIndex+1:])
			// Fields after the executable name begin at proc field 3. The
			// process start time is field 22, therefore index 19 here.
			if len(fields) > 19 && fields[19] != "" {
				return "proc:" + fields[19], nil
			}
		}
	}
	output, err := exec.Command("ps", "-o", "lstart=", "-p", strconv.Itoa(pid)).Output()
	if err != nil {
		return "", err
	}
	token := strings.Join(strings.Fields(string(output)), " ")
	if token == "" {
		return "", errors.New("empty process start token")
	}
	return "ps:" + token, nil
}

func artifactIdentity(work agenttask.WorkRecord) agenttask.ArtifactID {
	return agenttask.ArtifactID(
		digestStrings("artifact", string(work.Unit.ID), string(work.AttemptID)),
	)
}

func expandModelTarget(arguments []string, target string) []string {
	expanded := make([]string, len(arguments))
	for index, argument := range arguments {
		expanded[index] = strings.ReplaceAll(argument, "{{model}}", target)
	}
	return expanded
}

func replaceEnvironment(environment []string, key, value string) []string {
	prefix := key + "="
	out := make([]string, 0, len(environment)+1)
	for _, entry := range environment {
		if !strings.HasPrefix(entry, prefix) {
			out = append(out, entry)
		}
	}
	return append(out, prefix+value)
}

type nativeSessionReference struct {
	SchemaVersion       uint32 `json:"schema_version"`
	ProviderID          string `json:"provider_id"`
	ModelID             string `json:"model_id"`
	ProfileID           string `json:"profile_id"`
	ProfileRevision     string `json:"profile_revision"`
	SessionID           string `json:"session_id"`
	SessionDirectory    string `json:"session_directory"`
	ConfinementRevision string `json:"confinement_revision"`
	IsolationID         string `json:"isolation_id"`
	IsolationRevision   string `json:"isolation_revision"`
}

func prepareFreshCatalogCommand(
	resolved agentconfig.ResolvedProfile,
	target agenttask.ExecutionTarget,
	binding agenttask.ConfinementBinding,
	workingDirectory, prompt string,
	project agenttask.ProjectRecord,
	work agenttask.WorkRecord,
	purpose string,
) (agenttask.ConfinementCommand, *agenttask.LocatorRecord, error) {
	if err := validateCatalogTarget(resolved, target); err != nil {
		return agenttask.ConfinementCommand{}, nil, err
	}
	args := expandModelTarget(resolved.Profile.Args, resolved.Model.Target)
	var locator *agenttask.LocatorRecord
	if strings.EqualFold(resolved.Provider.ID, "pi") {
		reference, err := newNativeSessionReference(
			resolved,
			target,
			binding,
			project,
			work,
			purpose,
		)
		if err != nil {
			return agenttask.ConfinementCommand{}, nil, err
		}
		args = append(args,
			"--session-id", reference.SessionID,
			"--session-dir", reference.SessionDirectory,
		)
		encoded, err := json.Marshal(reference)
		if err != nil {
			return agenttask.ConfinementCommand{}, nil, err
		}
		record := agenttask.LocatorRecord{
			Kind:        agenttask.LocatorSession,
			Opaque:      string(encoded),
			Revision:    digestStrings("native-session-locator", string(encoded)),
			ProjectID:   project.ProjectID,
			WorkspaceID: project.WorkspaceID,
			WorkUnitID:  work.Unit.ID,
			AttemptID:   work.AttemptID,
		}
		locator = &record
	}
	args = append(args, prompt)
	environment, err := providerEnvironment(
		resolved,
		binding,
		workingDirectory,
	)
	if err != nil {
		return agenttask.ConfinementCommand{}, nil, err
	}
	return agenttask.ConfinementCommand{
		Name: resolved.Provider.Command,
		Args: args,
		Env:  environment,
	}, locator, nil
}

func prepareResumeCatalogCommand(
	resolved agentconfig.ResolvedProfile,
	target agenttask.ExecutionTarget,
	binding agenttask.ConfinementBinding,
	workingDirectory, prompt, nativeSessionFile string,
	reference nativeSessionReference,
) (agenttask.ConfinementCommand, error) {
	if err := validateCatalogTarget(resolved, target); err != nil {
		return agenttask.ConfinementCommand{}, err
	}
	if !strings.EqualFold(resolved.Provider.ID, "pi") {
		return agenttask.ConfinementCommand{}, errors.New(
			"taskloop: native session resume is supported only for Pi",
		)
	}
	if err := validateNativeSessionReference(reference, resolved, target, binding); err != nil {
		return agenttask.ConfinementCommand{}, err
	}
	if err := validateContainedRegularFile(
		reference.SessionDirectory,
		nativeSessionFile,
	); err != nil {
		return agenttask.ConfinementCommand{}, err
	}
	baseArgs := resolved.Profile.ResumeArgs
	if len(baseArgs) == 0 {
		baseArgs = resolved.Profile.Args
	}
	args := expandModelTarget(baseArgs, resolved.Model.Target)
	args = append(args,
		"--session", nativeSessionFile,
		"--session-dir", reference.SessionDirectory,
		prompt,
	)
	environment, err := providerEnvironment(
		resolved,
		binding,
		workingDirectory,
	)
	if err != nil {
		return agenttask.ConfinementCommand{}, err
	}
	return agenttask.ConfinementCommand{
		Name: resolved.Provider.Command,
		Args: args,
		Env:  environment,
	}, nil
}

func validateCatalogTarget(
	resolved agentconfig.ResolvedProfile,
	target agenttask.ExecutionTarget,
) error {
	if resolved.Provider.ID != target.ProviderID ||
		resolved.Model.ID != target.ModelID ||
		resolved.Profile.ID != target.ProfileID ||
		guardProfile(resolved).Revision != target.ProfileRevision {
		return errors.New("taskloop: selected provider identity does not match the catalog")
	}
	return nil
}

func providerEnvironment(
	resolved agentconfig.ResolvedProfile,
	binding agenttask.ConfinementBinding,
	workingDirectory string,
) ([]string, error) {
	environment := append([]string(nil), os.Environ()...)
	environment = append(environment, resolved.Profile.Env...)
	if workingDirectory != "" {
		environment = replaceEnvironment(environment, "PWD", workingDirectory)
	}
	if binding.TaskRoot != "" {
		environment = replaceEnvironment(
			environment,
			"IOP_AGENT_TASK_ROOT",
			binding.TaskRoot,
		)
	}
	if len(binding.WritableRoots) == 3 {
		tempRoot := binding.WritableRoots[1]
		for _, key := range []string{"TMPDIR", "TMP", "TEMP"} {
			environment = replaceEnvironment(environment, key, tempRoot)
		}
	}
	if !strings.EqualFold(resolved.Provider.ID, "codex") {
		return environment, nil
	}
	if len(binding.WritableRoots) != 3 {
		return nil, errors.New("taskloop: codex confinement has invalid writable roots")
	}
	codexHome, err := prepareCodexRuntimeHome(binding.WritableRoots[2])
	if err != nil {
		return nil, err
	}
	environment = replaceEnvironment(environment, "CODEX_HOME", codexHome)
	if runtime.GOOS == "darwin" {
		const systemCABundle = "/etc/ssl/cert.pem"
		if info, statErr := os.Stat(systemCABundle); statErr == nil && info.Mode().IsRegular() {
			environment = replaceEnvironment(
				environment,
				"SSL_CERT_FILE",
				systemCABundle,
			)
		}
	}
	return environment, nil
}

func prepareCodexRuntimeHome(cacheRoot string) (string, error) {
	if cacheRoot == "" || !filepath.IsAbs(cacheRoot) {
		return "", errors.New("taskloop: codex runtime cache root is invalid")
	}
	codexHome := filepath.Join(cacheRoot, "codex-home")
	if err := validateContainedPath(cacheRoot, codexHome); err != nil ||
		codexHome == cacheRoot {
		return "", errors.New("taskloop: codex runtime home escapes the cache root")
	}
	if err := os.MkdirAll(codexHome, 0o700); err != nil {
		return "", fmt.Errorf("taskloop: create codex runtime home: %w", err)
	}
	if err := os.Chmod(codexHome, 0o700); err != nil {
		return "", fmt.Errorf("taskloop: protect codex runtime home: %w", err)
	}
	sourceHome := strings.TrimSpace(os.Getenv("CODEX_HOME"))
	if sourceHome == "" {
		userHome, err := os.UserHomeDir()
		if err != nil {
			return "", fmt.Errorf("taskloop: resolve codex authentication home: %w", err)
		}
		sourceHome = filepath.Join(userHome, ".codex")
	}
	sourceHome, err := filepath.Abs(sourceHome)
	if err != nil {
		return "", fmt.Errorf("taskloop: resolve codex authentication home: %w", err)
	}
	for _, name := range []string{"auth.json", "config.toml"} {
		source := filepath.Join(sourceHome, name)
		info, statErr := os.Stat(source)
		if statErr != nil {
			if name == "config.toml" && os.IsNotExist(statErr) {
				continue
			}
			return "", fmt.Errorf(
				"taskloop: inspect codex authentication file %s: %w",
				name,
				statErr,
			)
		}
		if !info.Mode().IsRegular() {
			return "", fmt.Errorf(
				"taskloop: codex authentication file %s is not regular",
				name,
			)
		}
		destination := filepath.Join(codexHome, name)
		if existing, linkErr := os.Readlink(destination); linkErr == nil {
			if existing != source {
				return "", fmt.Errorf(
					"taskloop: codex runtime link %s changed identity",
					name,
				)
			}
			continue
		} else if !os.IsNotExist(linkErr) {
			return "", fmt.Errorf(
				"taskloop: inspect codex runtime link %s: %w",
				name,
				linkErr,
			)
		}
		if err := os.Symlink(source, destination); err != nil {
			return "", fmt.Errorf(
				"taskloop: project codex runtime link %s: %w",
				name,
				err,
			)
		}
	}
	return codexHome, nil
}

func newNativeSessionReference(
	resolved agentconfig.ResolvedProfile,
	target agenttask.ExecutionTarget,
	binding agenttask.ConfinementBinding,
	project agenttask.ProjectRecord,
	work agenttask.WorkRecord,
	purpose string,
) (nativeSessionReference, error) {
	if len(binding.WritableRoots) == 0 {
		return nativeSessionReference{}, errors.New(
			"taskloop: Pi native session requires a confined writable cache root",
		)
	}
	cacheRoot := binding.WritableRoots[len(binding.WritableRoots)-1]
	if !filepath.IsAbs(cacheRoot) || filepath.Clean(cacheRoot) != cacheRoot {
		return nativeSessionReference{}, errors.New(
			"taskloop: Pi native session cache root is invalid",
		)
	}
	sessionID := deterministicSessionID(
		string(project.ProjectID),
		string(project.WorkspaceID),
		string(work.Unit.ID),
		string(work.AttemptID),
		target.ProfileRevision,
		binding.Revision,
		purpose,
	)
	return nativeSessionReference{
		SchemaVersion:       nativeSessionLocatorSchemaVersion,
		ProviderID:          resolved.Provider.ID,
		ModelID:             resolved.Model.ID,
		ProfileID:           resolved.Profile.ID,
		ProfileRevision:     target.ProfileRevision,
		SessionID:           sessionID,
		SessionDirectory:    filepath.Join(cacheRoot, "pi-sessions", sessionID),
		ConfinementRevision: binding.Revision,
		IsolationID:         binding.IsolationID,
		IsolationRevision:   binding.IsolationRevision,
	}, nil
}

func deterministicSessionID(parts ...string) string {
	sum := sha256.Sum256([]byte(strings.Join(parts, "\x00")))
	sum[6] = (sum[6] & 0x0f) | 0x40
	sum[8] = (sum[8] & 0x3f) | 0x80
	return fmt.Sprintf(
		"%x-%x-%x-%x-%x",
		sum[0:4],
		sum[4:6],
		sum[6:8],
		sum[8:10],
		sum[10:16],
	)
}

func validateNativeSessionReference(
	reference nativeSessionReference,
	resolved agentconfig.ResolvedProfile,
	target agenttask.ExecutionTarget,
	binding agenttask.ConfinementBinding,
) error {
	if reference.SchemaVersion != nativeSessionLocatorSchemaVersion ||
		reference.ProviderID != resolved.Provider.ID ||
		reference.ModelID != resolved.Model.ID ||
		reference.ProfileID != resolved.Profile.ID ||
		reference.ProfileRevision != target.ProfileRevision ||
		reference.ConfinementRevision != binding.Revision ||
		reference.IsolationID != binding.IsolationID ||
		reference.IsolationRevision != binding.IsolationRevision ||
		reference.SessionID == "" {
		return errors.New("taskloop: native session locator identity mismatch")
	}
	if len(binding.WritableRoots) == 0 {
		return errors.New("taskloop: native session confinement has no writable roots")
	}
	cacheRoot := binding.WritableRoots[len(binding.WritableRoots)-1]
	if err := validateContainedPath(cacheRoot, reference.SessionDirectory); err != nil {
		return fmt.Errorf("taskloop: native session directory: %w", err)
	}
	if filepath.Base(reference.SessionDirectory) != reference.SessionID {
		return errors.New("taskloop: native session directory identity mismatch")
	}
	return nil
}

func resolveNativeSessionFile(reference nativeSessionReference) (string, error) {
	matches, err := filepath.Glob(
		filepath.Join(reference.SessionDirectory, "*"+reference.SessionID+"*.jsonl"),
	)
	if err != nil {
		return "", err
	}
	if len(matches) != 1 {
		return "", errors.New(
			"taskloop: Pi repair requires exactly one matching native session file",
		)
	}
	if err := validateContainedRegularFile(reference.SessionDirectory, matches[0]); err != nil {
		return "", err
	}
	return matches[0], nil
}

func validateContainedPath(root, path string) error {
	if !filepath.IsAbs(root) || !filepath.IsAbs(path) ||
		filepath.Clean(root) != root || filepath.Clean(path) != path {
		return errors.New("path must be absolute and clean")
	}
	relative, err := filepath.Rel(root, path)
	if err != nil || relative == ".." ||
		strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
		return errors.New("path escapes its owned root")
	}
	return nil
}

func validateContainedRegularFile(root, path string) error {
	if err := validateContainedPath(root, path); err != nil {
		return err
	}
	info, err := os.Lstat(path)
	if err != nil {
		return err
	}
	if !info.Mode().IsRegular() || info.Size() <= 0 ||
		info.Size() > maxWorkflowArtifactBytes {
		return errors.New("taskloop: native session is not one bounded regular file")
	}
	return nil
}

func invokeConfined(
	ctx context.Context,
	confinement agenttask.InvocationConfinement,
	command agenttask.ConfinementCommand,
) error {
	if confinement == nil {
		return errors.New("taskloop: executable confinement is unavailable")
	}
	binding := confinement.Binding()
	if confinement.Revision() != binding.Revision {
		return errors.New("taskloop: executable confinement revision mismatch")
	}
	if err := confinement.Validate(binding); err != nil {
		return fmt.Errorf("taskloop: validate executable confinement: %w", err)
	}
	started, err := confinement.Start(ctx, command)
	if err != nil {
		return err
	}
	if started == nil || started.Child() == nil || started.Child().Process == nil ||
		started.Stdin() == nil || started.Stdout() == nil || started.Stderr() == nil {
		if started != nil {
			_ = started.Abort()
		}
		return errors.New("taskloop: confined provider child is incomplete")
	}
	_ = started.Stdin().Close()
	var stdout, stderr boundedBuffer
	var drains sync.WaitGroup
	drains.Add(2)
	go func() {
		defer drains.Done()
		_, _ = io.Copy(&stdout, started.Stdout())
		_ = started.Stdout().Close()
	}()
	go func() {
		defer drains.Done()
		_, _ = io.Copy(&stderr, started.Stderr())
		_ = started.Stderr().Close()
	}()
	done := make(chan error, 1)
	go func() {
		waitErr := started.Child().Wait()
		drains.Wait()
		done <- waitErr
	}()
	select {
	case <-ctx.Done():
		if signalErr := started.Child().Process.Signal(syscall.SIGTERM); signalErr != nil &&
			!errors.Is(signalErr, os.ErrProcessDone) {
			_ = started.Child().Process.Kill()
		}
		<-done
		return ctx.Err()
	case waitErr := <-done:
		if waitErr != nil {
			return errors.New("taskloop: confined provider exited unsuccessfully")
		}
		return nil
	}
}