iop/apps/edge/internal/service/single_request_observation.go

package service

import (
	"crypto/rand"
	"encoding/hex"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// singleRequestEventClass is the closed top-level event class for every
// single-request observation. It scopes the lifecycle without exposing request
// or stage identity (SDD S07).
type singleRequestEventClass string

const (
	singleRequestEventClassRequest  singleRequestEventClass = "request"
	singleRequestEventClassStage    singleRequestEventClass = "stage"
	singleRequestEventClassTool     singleRequestEventClass = "tool"
	singleRequestEventClassCleanup  singleRequestEventClass = "cleanup"
	singleRequestEventClassTerminal singleRequestEventClass = "terminal"
)

// singleRequestStage is the closed stage role observed on stage events.
type singleRequestStage string

const (
	singleRequestStagePlan   singleRequestStage = "plan"
	singleRequestStageWork   singleRequestStage = "work"
	singleRequestStageReview singleRequestStage = "review"
)

// singleRequestOperation is the closed operation observed on request/tool/cleanup events.
type singleRequestOperation string

const (
	singleRequestOperationPlan     singleRequestOperation = "plan"
	singleRequestOperationWork     singleRequestOperation = "work"
	singleRequestOperationReview   singleRequestOperation = "review"
	singleRequestOperationTool     singleRequestOperation = "tool"
	singleRequestOperationCleanup  singleRequestOperation = "cleanup"
	singleRequestOperationTerminal singleRequestOperation = "terminal"
	singleRequestOperationTotal    singleRequestOperation = "total"
)

// singleRequestOutcome is the closed outcome observed on stage/terminal events.
type singleRequestOutcome string

const (
	singleRequestOutcomeSuccess singleRequestOutcome = "success"
	singleRequestOutcomeError   singleRequestOutcome = "error"
	singleRequestOutcomeCancel  singleRequestOutcome = "cancel"
)

// singleRequestErrorClass is the closed error classification observed on
// stage/terminal events when outcome is error or cancel. It never carries
// raw error text.
type singleRequestErrorClass string

const (
	singleRequestErrorClassProvider           singleRequestErrorClass = "provider"
	singleRequestErrorClassValidation         singleRequestErrorClass = "validation"
	singleRequestErrorClassTimeout            singleRequestErrorClass = "timeout"
	singleRequestErrorClassCancel             singleRequestErrorClass = "cancel"
	singleRequestErrorClassInternalToolBudget singleRequestErrorClass = "internal_tool_budget"
	singleRequestErrorClassInternalToolFailed singleRequestErrorClass = "internal_tool_failed"
	singleRequestErrorClassWorkspaceCleanup   singleRequestErrorClass = "workspace_cleanup"
)

// singleRequestDTO is the closed, copy-safe single-request observation record.
// It contains only closed identities, durations/counts, and truncated booleans.
// It never contains request text, public model, provider id, Node/root/path,
// command/template/env, tool input/output, error string, header, credential,
// or raw terminal output.
type singleRequestDTO struct {
	// EventClass is the closed top-level event class.
	EventClass singleRequestEventClass
	// Stage is the closed stage role (plan/work/review). Empty for non-stage events.
	Stage singleRequestStage
	// Operation is the closed operation observed on request/tool/cleanup events.
	Operation singleRequestOperation
	// Outcome is the closed outcome (success/error/cancel).
	Outcome singleRequestOutcome
	// ErrorClass is the closed error classification. Empty when outcome is success.
	ErrorClass singleRequestErrorClass
	// DurationMS is the duration in milliseconds for stage/tool/cleanup/total events.
	DurationMS int64
	// ToolCount is the number of tool calls during a stage. Zero for non-stage events.
	ToolCount int
	// HasResult is true when a finalizing candidate was prepared. Truncated boolean.
	HasResult bool
	// Correlation is a bounded generated execution correlation id for later logs.
	Correlation string
}

// singleRequestDTOIsValid reports whether d has a valid event class.
func singleRequestDTOIsValid(d singleRequestDTO) bool {
	return singleRequestEventClassIsValid(d.EventClass)
}

// singleRequestEventClassIsValid reports whether c is a known event class.
func singleRequestEventClassIsValid(c singleRequestEventClass) bool {
	switch c {
	case singleRequestEventClassRequest, singleRequestEventClassStage,
		singleRequestEventClassTool, singleRequestEventClassCleanup,
		singleRequestEventClassTerminal:
		return true
	default:
		return false
	}
}

// singleRequestStageIsValid reports whether s is a known stage role.
func singleRequestStageIsValid(s singleRequestStage) bool {
	switch s {
	case singleRequestStagePlan, singleRequestStageWork, singleRequestStageReview:
		return true
	default:
		return false
	}
}

// singleRequestOperationIsValid reports whether o is a known operation.
func singleRequestOperationIsValid(o singleRequestOperation) bool {
	switch o {
	case singleRequestOperationPlan, singleRequestOperationWork, singleRequestOperationReview,
		singleRequestOperationTool, singleRequestOperationCleanup,
		singleRequestOperationTerminal, singleRequestOperationTotal:
		return true
	default:
		return false
	}
}

// singleRequestOutcomeIsValid reports whether o is a known outcome.
func singleRequestOutcomeIsValid(o singleRequestOutcome) bool {
	switch o {
	case singleRequestOutcomeSuccess, singleRequestOutcomeError, singleRequestOutcomeCancel:
		return true
	default:
		return false
	}
}

// singleRequestErrorClassIsValid reports whether e is a known error class.
func singleRequestErrorClassIsValid(e singleRequestErrorClass) bool {
	switch e {
	case singleRequestErrorClassProvider, singleRequestErrorClassValidation,
		singleRequestErrorClassTimeout, singleRequestErrorClassCancel,
		singleRequestErrorClassInternalToolBudget, singleRequestErrorClassInternalToolFailed,
		singleRequestErrorClassWorkspaceCleanup:
		return true
	default:
		return false
	}
}

// singleRequestNormalizeStage converts a raw stage string to its closed form.
// Unknown values become empty so callers cannot smuggle arbitrary text.
func singleRequestNormalizeStage(raw string) singleRequestStage {
	switch singleRequestStage(raw) {
	case singleRequestStagePlan, singleRequestStageWork, singleRequestStageReview:
		return singleRequestStage(raw)
	default:
		return ""
	}
}

// singleRequestNormalizeOperation converts a raw operation string to its closed form.
// Unknown values become empty.
func singleRequestNormalizeOperation(raw string) singleRequestOperation {
	switch singleRequestOperation(raw) {
	case singleRequestOperationPlan, singleRequestOperationWork, singleRequestOperationReview,
		singleRequestOperationTool, singleRequestOperationCleanup,
		singleRequestOperationTerminal, singleRequestOperationTotal:
		return singleRequestOperation(raw)
	default:
		return ""
	}
}

// singleRequestNormalizeOutcome converts a raw outcome string to its closed form.
// Unknown values become empty.
func singleRequestNormalizeOutcome(raw string) singleRequestOutcome {
	switch singleRequestOutcome(raw) {
	case singleRequestOutcomeSuccess, singleRequestOutcomeError, singleRequestOutcomeCancel:
		return singleRequestOutcome(raw)
	default:
		return ""
	}
}

// singleRequestNormalizeErrorClass converts a raw error class string to its closed form.
// Unknown values become empty.
func singleRequestNormalizeErrorClass(raw string) singleRequestErrorClass {
	switch singleRequestErrorClass(raw) {
	case singleRequestErrorClassProvider, singleRequestErrorClassValidation,
		singleRequestErrorClassTimeout, singleRequestErrorClassCancel,
		singleRequestErrorClassInternalToolBudget, singleRequestErrorClassInternalToolFailed,
		singleRequestErrorClassWorkspaceCleanup:
		return singleRequestErrorClass(raw)
	default:
		return ""
	}
}

// singleRequestContainsSecretSentinel reports whether s contains secret sentinels.
func singleRequestContainsSecretSentinel(s string) bool {
	lower := strings.ToLower(s)
	return strings.Contains(lower, "secret") ||
		strings.Contains(lower, "bearer") ||
		strings.Contains(lower, "api_key") ||
		strings.Contains(lower, "token") ||
		strings.Contains(s, "\x00")
}

// singleRequestSanitizeString truncates strings that contain secret sentinels
// or exceed the bounded length. Returns empty for secret-containing strings.
func singleRequestSanitizeString(s string) string {
	if singleRequestContainsSecretSentinel(s) {
		return ""
	}
	if len(s) > 64 {
		return s[:64]
	}
	return s
}

var singleRequestCorrelationFallback atomic.Uint64

// newSingleRequestCorrelationID creates one bounded, raw-input-independent
// correlation id. It is generated once per accumulator rather than derived
// from caller request or stage identifiers, which may contain sensitive input.
func newSingleRequestCorrelationID() string {
	var bytes [16]byte
	if _, err := rand.Read(bytes[:]); err == nil {
		return "sr-" + hex.EncodeToString(bytes[:])
	}
	return "sr-fallback-" + strconv.FormatUint(singleRequestCorrelationFallback.Add(1), 36)
}

// singleRequestObserver is the service-owned observation contract. Implementations
// own storage and retention; callers only own the bounded DTO inputs.
// Emit must not block indefinitely — sinks that need bounded work should apply
// their own timeout internally.
type singleRequestObserver interface {
	Emit(dto singleRequestDTO) error
}

// singleRequestNoopObserver discards every observation. It is the default
// observer for hosts that have not wired a logging backend yet.
type singleRequestNoopObserver struct{}

// Emit discards the observation and always returns nil.
func (singleRequestNoopObserver) Emit(_ singleRequestDTO) error {
	return nil
}

// singleRequestObserverFailureHook is called when an observer failure occurs.
// It is optional; the observer isolates failures so they never affect request
// results.
type singleRequestObserverFailureHook func(dto singleRequestDTO, err error)

// singleRequestSafeObserver wraps an inner observer with failure isolation.
// If the inner observer panics or returns an error, the failure is reported
// through the hook (which is also panic-isolated) and Emit returns nil.
// Both observer and hook panics are completely isolated so the request path
// is never interrupted.
type singleRequestSafeObserver struct {
	inner     singleRequestObserver
	onFailure singleRequestObserverFailureHook
	failures  int64
	mu        sync.Mutex
}

// Emit forwards the DTO to the inner observer with failure isolation.
// If the inner observer returns an error or panics, the failure is reported
// through the hook (which is also panic-isolated) and Emit returns nil.
func (s *singleRequestSafeObserver) Emit(dto singleRequestDTO) error {
	if s == nil || s.inner == nil {
		return nil
	}
	func() {
		defer func() {
			if r := recover(); r != nil {
				s.mu.Lock()
				s.failures++
				s.mu.Unlock()
				if s.onFailure != nil {
					func() {
						defer func() {
							_ = recover()
						}()
						s.onFailure(dto, errObserverPanic(r))
					}()
				}
			}
		}()
		if err := s.inner.Emit(dto); err != nil {
			s.mu.Lock()
			s.failures++
			s.mu.Unlock()
			if s.onFailure != nil {
				func() {
					defer func() {
						_ = recover()
					}()
					s.onFailure(dto, err)
				}()
			}
			return
		}
	}()
	return nil
}

// failureCount returns the number of isolated failures observed so far.
// Safe for concurrent reads from tests.
func (s *singleRequestSafeObserver) failureCount() int64 {
	if s == nil {
		return 0
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.failures
}

// errObserverPanic wraps a recovered panic value into a sentinel error.
// The hook receives this error to distinguish panic vs. Emit error.
func errObserverPanic(r any) error {
	return errSingleRequestObserverPanic{reason: r}
}

type errSingleRequestObserverPanic struct {
	reason any
}

func (e errSingleRequestObserverPanic) Error() string {
	return "single-request observer panic"
}

// singleRequestClock is the injectable clock interface for deterministic testing.
// The production implementation delegates to time.Now and time.Since.
type singleRequestClock interface {
	Now() time.Time
	Since(time.Time) time.Duration
}

// singleRequestRealClock is the production clock implementation.
type singleRequestRealClock struct{}

// Now returns the current wall-clock time.
func (singleRequestRealClock) Now() time.Time {
	return time.Now()
}

// Since returns the duration since t.
func (singleRequestRealClock) Since(t time.Time) time.Duration {
	return time.Since(t)
}

// singleRequestManualClock is the deterministic clock for tests. It advances
// only when Advance is called, allowing precise timing assertions without
// real elapsed time.
type singleRequestManualClock struct {
	mu      sync.Mutex
	now     time.Time
	advance time.Duration
}

// newSingleRequestManualClock creates a manual clock starting at the given time.
func newSingleRequestManualClock(start time.Time) *singleRequestManualClock {
	return &singleRequestManualClock{now: start}
}

// Now returns the current manual clock time.
func (c *singleRequestManualClock) Now() time.Time {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.now
}

// Since returns the duration since t using the manual clock.
func (c *singleRequestManualClock) Since(t time.Time) time.Duration {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.now.Sub(t)
}

// Advance advances the manual clock by d. Safe for concurrent use.
func (c *singleRequestManualClock) Advance(d time.Duration) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.now = c.now.Add(d)
}

// singleRequestTimingAccumulator accumulates provider-active stage time,
// tool time, cleanup time, and request total time. It is copy-safe and
// thread-safe.
type singleRequestTimingAccumulator struct {
	mu        sync.Mutex
	startTime time.Time
	clock     singleRequestClock
	observer  *singleRequestSafeObserver
	// correlation joins every lifecycle DTO for this request without carrying
	// any caller-controlled identifier.
	correlation string

	// Provider-active stage time: time spent in plan/work/review stages.
	// Pauses during internal_tool execution.
	stageActiveStart time.Time
	stageActiveMs    int64

	// Tool time: time spent in internal_tool execution.
	toolStart time.Time
	toolMs    int64

	// Cleanup time: time spent in cleanup.
	cleanupStart time.Time
	cleanupMs    int64

	// Total time: measured from request start to terminal resolution.
	totalMs int64

	// Terminal outcome and error class, set exactly once by the terminal winner.
	terminalOutcome    singleRequestOutcome
	terminalErrorClass singleRequestErrorClass
	terminalHasResult  bool

	// Stage event count per stage role.
	stageEventCount int

	// Tool call count during active stage.
	toolCallCount int

	// pendingStageDuration accumulates stage time between tool enter/exit pairs.
	// It is added to stageActiveMs on stage exit.
	pendingStageDurationMs int64
	activeStage            singleRequestStage
	stageActive            bool
	pendingStageClose      *singleRequestPendingStageClose
}

type singleRequestPendingStageClose struct {
	stage      singleRequestStage
	outcome    singleRequestOutcome
	errorClass singleRequestErrorClass
}

// newSingleRequestTimingAccumulator creates a new timing accumulator.
// The observer is wrapped in a safe observer for failure isolation.
func newSingleRequestTimingAccumulator(clock singleRequestClock, observer singleRequestObserver) *singleRequestTimingAccumulator {
	if clock == nil {
		clock = singleRequestRealClock{}
	}
	if observer == nil {
		observer = singleRequestNoopObserver{}
	}
	safe := &singleRequestSafeObserver{inner: observer}
	return &singleRequestTimingAccumulator{
		startTime:   clock.Now(),
		clock:       clock,
		observer:    safe,
		correlation: newSingleRequestCorrelationID(),
	}
}

// onStageEnter records the start of a provider-active stage (plan/work/review).
// It resets the pending stage duration accumulator.
func (a *singleRequestTimingAccumulator) onStageEnter(stage ...singleRequestStage) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if len(stage) > 0 && stage[0] != "" {
		if a.activeStage == stage[0] {
			return
		}
		a.activeStage = stage[0]
	}
	a.stageActiveStart = a.clock.Now()
	a.pendingStageDurationMs = 0
	a.stageActive = true
}

// onStageExit records the end of a provider-active stage and emits a stage event.
// It includes all accumulated stage time (excluding tool time).
func (a *singleRequestTimingAccumulator) onStageExit(stage singleRequestStage, outcome singleRequestOutcome, errorClass singleRequestErrorClass) {
	a.mu.Lock()
	if stage == "" || a.activeStage != "" && a.activeStage != stage {
		a.mu.Unlock()
		return
	}
	if !a.toolStart.IsZero() {
		// A terminal can win while a Node tool is still settling. Keep the
		// semantic stage open until the tool has emitted and contributed to its
		// count, then emit the stage without restarting its active timer.
		if a.pendingStageClose == nil {
			a.pendingStageClose = &singleRequestPendingStageClose{stage: stage, outcome: outcome, errorClass: errorClass}
		}
		a.mu.Unlock()
		return
	}
	dto := a.closeStageLocked(stage, outcome, errorClass)
	a.mu.Unlock()
	a.emitSafe(dto)
}

// closeStageLocked finalizes the active semantic stage. Caller must hold a.mu.
func (a *singleRequestTimingAccumulator) closeStageLocked(stage singleRequestStage, outcome singleRequestOutcome, errorClass singleRequestErrorClass) singleRequestDTO {
	duration := a.pendingStageDurationMs
	if !a.stageActiveStart.IsZero() {
		duration += a.clock.Since(a.stageActiveStart).Milliseconds()
	}
	a.stageActiveMs += duration
	a.stageActiveStart = time.Time{}
	a.pendingStageDurationMs = 0
	a.stageEventCount++
	toolCount := a.toolCallCount
	a.toolCallCount = 0
	a.activeStage = ""
	a.stageActive = false
	a.pendingStageClose = nil
	return singleRequestDTO{
		EventClass:  singleRequestEventClassStage,
		Stage:       stage,
		Operation:   singleRequestNormalizeOperation(string(stage)),
		Outcome:     outcome,
		ErrorClass:  errorClass,
		DurationMS:  duration,
		ToolCount:   toolCount,
		Correlation: a.correlation,
	}
}

// onToolEnter records the start of internal_tool execution and pauses stage timing.
// The elapsed stage time is accumulated in pendingStageDurationMs.
func (a *singleRequestTimingAccumulator) onToolEnter() {
	a.mu.Lock()
	defer a.mu.Unlock()
	if !a.stageActiveStart.IsZero() {
		a.pendingStageDurationMs += a.clock.Since(a.stageActiveStart).Milliseconds()
		a.stageActiveStart = time.Time{}
	}
	if a.toolStart.IsZero() {
		a.toolStart = a.clock.Now()
	}
}

// onToolExit records one actual Node tool outcome, emits its closed DTO, and
// resumes the still-active semantic provider stage.
func (a *singleRequestTimingAccumulator) onToolExit(outcome singleRequestOutcome, errorClass singleRequestErrorClass) {
	a.mu.Lock()
	if a.toolStart.IsZero() {
		a.mu.Unlock()
		return
	}
	duration := a.clock.Since(a.toolStart).Milliseconds()
	a.toolMs += duration
	a.toolStart = time.Time{}
	a.toolCallCount++
	toolDTO := singleRequestDTO{
		EventClass:  singleRequestEventClassTool,
		Operation:   singleRequestOperationTool,
		Outcome:     outcome,
		ErrorClass:  errorClass,
		DurationMS:  duration,
		Correlation: a.correlation,
	}
	var stageDTO *singleRequestDTO
	if pending := a.pendingStageClose; pending != nil {
		dto := a.closeStageLocked(pending.stage, pending.outcome, pending.errorClass)
		stageDTO = &dto
	} else if a.stageActive {
		// Resume only a still-active semantic stage. A terminal stage close must
		// never leave a phantom active timer behind.
		a.stageActiveStart = a.clock.Now()
	}
	a.mu.Unlock()

	a.emitSafe(toolDTO)
	if stageDTO != nil {
		a.emitSafe(*stageDTO)
	}
}

// onCleanupEnter records the start of cleanup.
func (a *singleRequestTimingAccumulator) onCleanupEnter() {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.cleanupStart = a.clock.Now()
}

// onCleanupExit records the end of cleanup and emits a cleanup event.
func (a *singleRequestTimingAccumulator) onCleanupExit(outcome singleRequestOutcome, errorClass singleRequestErrorClass) {
	a.mu.Lock()
	if a.cleanupStart.IsZero() {
		a.mu.Unlock()
		return
	}
	duration := a.clock.Since(a.cleanupStart).Milliseconds()
	a.cleanupMs += duration
	a.cleanupStart = time.Time{}
	a.mu.Unlock()

	dto := singleRequestDTO{
		EventClass:  singleRequestEventClassCleanup,
		Operation:   singleRequestOperationCleanup,
		Outcome:     outcome,
		ErrorClass:  errorClass,
		DurationMS:  duration,
		Correlation: a.correlation,
	}
	a.emitSafe(dto)
}

// onTerminal records the terminal outcome exactly once and emits a terminal event.
// The terminal winner owns exactly one terminal event and one request-total event.
func (a *singleRequestTimingAccumulator) onTerminal(outcome singleRequestOutcome, errorClass singleRequestErrorClass, hasResult bool) {
	a.mu.Lock()
	if a.terminalOutcome != "" {
		// Already recorded by another caller; ignore.
		a.mu.Unlock()
		return
	}
	a.terminalOutcome = outcome
	a.terminalErrorClass = errorClass
	a.terminalHasResult = hasResult
	a.totalMs = a.clock.Since(a.startTime).Milliseconds()
	a.mu.Unlock()

	dto := singleRequestDTO{
		EventClass:  singleRequestEventClassTerminal,
		Operation:   singleRequestOperationTerminal,
		Outcome:     outcome,
		ErrorClass:  errorClass,
		DurationMS:  a.totalMs,
		HasResult:   hasResult,
		Correlation: a.correlation,
	}
	a.emitSafe(dto)
}

// onRequest records the initial request event.
func (a *singleRequestTimingAccumulator) onRequest() {
	dto := singleRequestDTO{
		EventClass:  singleRequestEventClassRequest,
		Operation:   singleRequestOperationTotal,
		Outcome:     singleRequestOutcomeSuccess,
		Correlation: a.correlation,
	}
	a.emitSafe(dto)
}

// emitSafe emits a DTO through the safe observer. Observer failures are isolated
// and never propagate to the caller.
func (a *singleRequestTimingAccumulator) emitSafe(dto singleRequestDTO) {
	if !singleRequestDTOIsValid(dto) {
		return
	}
	// Normalize any non-empty string fields to closed form.
	dto.Stage = singleRequestNormalizeStage(string(dto.Stage))
	dto.Operation = singleRequestNormalizeOperation(string(dto.Operation))
	dto.Outcome = singleRequestNormalizeOutcome(string(dto.Outcome))
	dto.ErrorClass = singleRequestNormalizeErrorClass(string(dto.ErrorClass))
	// Sanitize correlation id.
	dto.Correlation = singleRequestSanitizeString(string(dto.Correlation))
	a.observer.Emit(dto)
}

// timingSnapshot returns a copy-safe snapshot of the accumulated timing data.
// Used for verification in tests.
func (a *singleRequestTimingAccumulator) timingSnapshot() singleRequestTimingSnapshot {
	a.mu.Lock()
	defer a.mu.Unlock()
	return singleRequestTimingSnapshot{
		StageActiveMs:      a.stageActiveMs,
		ToolMs:             a.toolMs,
		CleanupMs:          a.cleanupMs,
		TotalMs:            a.totalMs,
		StageEventCount:    a.stageEventCount,
		ToolCallCount:      a.toolCallCount,
		TerminalOutcome:    a.terminalOutcome,
		TerminalErrorClass: a.terminalErrorClass,
		TerminalHasResult:  a.terminalHasResult,
	}
}

// singleRequestTimingSnapshot is a copy-safe snapshot of accumulated timing data.
type singleRequestTimingSnapshot struct {
	StageActiveMs      int64
	ToolMs             int64
	CleanupMs          int64
	TotalMs            int64
	StageEventCount    int
	ToolCallCount      int
	TerminalOutcome    singleRequestOutcome
	TerminalErrorClass singleRequestErrorClass
	TerminalHasResult  bool
}