tutti/packages/agent/host/command_failure.go

package agenthost

import (
	"context"
	"strings"
	"sync"
	"time"
)

// commandPreconditionStage names the boundary emission for a command that
// failed before it reached any observable lifecycle step: argument validation,
// submit-claim preparation, send fences, and lock acquisition.
const commandPreconditionStage = "precondition"

// commandTerminalFailure aggregates one Host command into at most one
// TerminalFailure. LifecycleStep stays diagnostic; it only records the first
// failing stage so the command boundary can name where the command broke.
// A more specific emission inside the command (guidance target, goal control,
// durable delta) marks the aggregation as already reported.
type commandTerminalFailure struct {
	flow string

	mu                            sync.Mutex
	workspaceID                   string
	agentSessionID                string
	operationID                   string
	requestID                     string
	clientSubmitID                string
	turnID                        string
	provider                      string
	providerAcceptanceDiagnostics *RuntimeProviderAcceptanceDiagnostics
	stage                         string
	startedAt                     time.Time
	emitted                       bool
}

type commandTerminalFailureInput struct {
	flow           string
	workspaceID    string
	agentSessionID string
	operationID    string
	requestID      string
	clientSubmitID string
	turnID         string
}

type commandTerminalFailureKey struct{}

// beginCommand scopes a terminal-failure aggregation to one Host command. A
// nested command scopes its own aggregation for its callees only.
func (h *Host) beginCommand(
	ctx context.Context,
	input commandTerminalFailureInput,
) (context.Context, *commandTerminalFailure) {
	command := &commandTerminalFailure{
		flow:           strings.TrimSpace(input.flow),
		workspaceID:    strings.TrimSpace(input.workspaceID),
		agentSessionID: strings.TrimSpace(input.agentSessionID),
		operationID:    strings.TrimSpace(input.operationID),
		requestID:      strings.TrimSpace(input.requestID),
		clientSubmitID: strings.TrimSpace(input.clientSubmitID),
		turnID:         strings.TrimSpace(input.turnID),
	}
	if h != nil {
		command.startedAt = h.now()
	}
	if h == nil {
		return ctx, command
	}
	return context.WithValue(ctx, commandTerminalFailureKey{}, command), command
}

func commandTerminalFailureFrom(ctx context.Context) *commandTerminalFailure {
	command, _ := ctx.Value(commandTerminalFailureKey{}).(*commandTerminalFailure)
	return command
}

// recordCommandFailureStage keeps the first failing stage of the running
// command. Steps from a different flow (notably session_create_cleanup, which
// runs only after a primary failure) never claim the stage.
func recordCommandFailureStage(ctx context.Context, flow, workspaceID, agentSessionID, provider, stage string, err error) {
	command := commandTerminalFailureFrom(ctx)
	if command == nil || err == nil || command.flow != flow {
		return
	}
	command.mu.Lock()
	defer command.mu.Unlock()
	if command.stage != "" {
		return
	}
	command.stage = stage
	if value := strings.TrimSpace(workspaceID); value != "" {
		command.workspaceID = value
	}
	if value := strings.TrimSpace(agentSessionID); value != "" {
		command.agentSessionID = value
	}
	if value := strings.TrimSpace(provider); value != "" {
		command.provider = value
	}
}

func markCommandTerminalFailureEmitted(ctx context.Context) {
	command := commandTerminalFailureFrom(ctx)
	if command == nil {
		return
	}
	command.mu.Lock()
	command.emitted = true
	command.mu.Unlock()
}

func recordProviderAcceptanceDiagnostics(
	ctx context.Context,
	dispatch RuntimeProviderDispatchResult,
) {
	command := commandTerminalFailureFrom(ctx)
	if command == nil || dispatch.AcceptanceDiagnostics == nil {
		return
	}
	diagnostics := *dispatch.AcceptanceDiagnostics
	command.mu.Lock()
	command.providerAcceptanceDiagnostics = &diagnostics
	command.mu.Unlock()
}

// finish emits the single aggregated failure for a command that returned err.
func (c *commandTerminalFailure) finish(ctx context.Context, h *Host, err error) {
	if c == nil || err == nil {
		return
	}
	c.mu.Lock()
	emitted, stage := c.emitted, c.stage
	failure := TerminalFailure{
		Flow:                          c.flow,
		WorkspaceID:                   c.workspaceID,
		AgentSessionID:                c.agentSessionID,
		OperationID:                   c.operationID,
		RequestID:                     c.requestID,
		ClientSubmitID:                c.clientSubmitID,
		TurnID:                        c.turnID,
		Provider:                      c.provider,
		ProviderAcceptanceDiagnostics: c.providerAcceptanceDiagnostics,
	}
	startedAt := c.startedAt
	c.mu.Unlock()
	if emitted {
		return
	}
	if stage == "" {
		stage = commandPreconditionStage
	}
	failure.FailureStage = stage
	failure.ErrorCode = terminalFailureCode(err)
	failure.ErrorMessage = err.Error()
	failure.Retryable = isRetryableRuntimeOperationError(err)
	if duration := h.now().Sub(startedAt).Milliseconds(); duration > 0 {
		failure.DurationMS = duration
	}
	h.observeTerminalFailure(ctx, failure)
}