story/internal/orchestration/hardware_agent.go

package orchestration

import (
	"context"
	"crypto/sha256"
	"crypto/subtle"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"strings"
	"time"

	"data-terminal-cloud/internal/audit"
	"data-terminal-cloud/internal/conversation"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/idempotency"
	"data-terminal-cloud/internal/outbox"

	"github.com/jackc/pgx/v5"
)

var (
	ErrHardwareSessionProof  = errors.New("hardware agent session proof is invalid")
	ErrHardwareEventConflict = errors.New("hardware agent event conflicts with stored event")
	ErrCallContractRequired  = errors.New("CALL-MEMORY-1.1 required after a critical memory change")
)

func (s *Service) CreateHardwareAgent(ctx context.Context, identity deviceauth.Identity, idempotencyKey, requestID string, rawRequest []byte, input HardwareAgentCreateInput) (HardwareAgentCreateResponse, bool, error) {
	tenantID := identity.TenantID
	request := idempotency.Request{TenantID: &tenantID, ActorKey: "device:" + identity.PublicID,
		Scope: "POST /device/v3/agent-sessions", Key: idempotencyKey,
		RequestHash: idempotency.Hash(rawRequest), ExpiresAfter: 24 * time.Hour}
	memoryVector, memoryPolicy, err := s.prepareMemoryQuery(ctx, identity, input.RequestedGoal)
	if err != nil {
		return HardwareAgentCreateResponse{}, false, err
	}
	var vikingSubject, vikingContext string
	var vikingNativeMemory *HardwareAgentMemoryConfig
	if s.viking != nil {
		readContext, cancel := context.WithTimeout(ctx, 3*time.Second)
		native, readErr := s.viking.NativeAccessForDevice(readContext, identity.TenantID, identity.ID)
		cancel()
		if readErr != nil {
			slog.WarnContext(ctx, "Viking native memory configuration unavailable", "error", readErr)
		} else if native != nil {
			vikingSubject = native.SubjectID
			vikingNativeMemory = &HardwareAgentMemoryConfig{
				CollectionName: native.CollectionName, UserID: native.UserID, AssistantID: native.AssistantID,
				MemoryTypes: native.MemoryTypes, Limit: native.Limit, TransitionWords: native.TransitionWords,
			}
		} else {
			// Older deployments without a native collection name keep the
			// platform-side startup retrieval path as a compatibility fallback.
			fallbackContext, fallbackCancel := context.WithTimeout(ctx, 3*time.Second)
			vikingSubject, vikingContext, readErr = s.viking.PrepareForDevice(fallbackContext, identity.TenantID, identity.ID)
			fallbackCancel()
			if readErr != nil {
				slog.WarnContext(ctx, "Viking startup memory unavailable", "error", readErr)
			}
		}
	}
	result, err := s.executor.ExecuteGuarded(ctx, request, onboardingCreateGuard(identity, input), func(operationContext context.Context, tx pgx.Tx) (idempotency.Result, error) {
		now := s.now().UTC()
		created, internalID, createErr := s.repository.Create(operationContext, tx, identity, CreateInput{
			DeviceTime: input.DeviceTime, MonotonicMS: input.MonotonicMS, Network: input.Network,
			RequestedGoal: input.RequestedGoal, ConfigRevision: input.ConfigRevision,
			Capabilities: Capabilities{UploadResume: boolPointer(input.Capabilities.TranscriptUpload)},
		}, AllocationOptions{Now: now, AllocationExpires: now.Add(s.options.AllocationTTL),
			TokenExpires: now.Add(s.options.TokenTTL), RTCAppID: s.options.RTCAppID,
			VoiceProvider: s.options.VoiceProvider, EngineMode: s.options.EngineMode,
			HeartbeatInterval: s.options.HeartbeatInterval, MemoryQueryVector: memoryVector,
			VikingSubjectID: vikingSubject, VikingContext: vikingContext, VikingNativeMemory: vikingNativeMemory,
			MemoryPolicy: memoryPolicy, RequirePublishedRouting: s.options.RequirePublishedRouting,
			RuntimeType: RuntimeVolcHardwareAgent, HardwareBotID: s.options.HardwareBotID,
			HardwareTransport: s.options.HardwareTransport, HardwareCapabilities: input.Capabilities,
			SubtitleCallbackURL: s.options.SubtitleCallbackURL, SubtitleCallbackSecret: s.options.SubtitleCallbackSecret,
			SubtitlePrimary:       s.options.SubtitlePrimary,
			HardwareMemoryToolURL: s.hardwareMemoryToolURL})
		if createErr != nil {
			return idempotency.Result{}, createErr
		}
		if err := attachOnboardingConversation(operationContext, tx, identity, input, internalID); err != nil {
			return idempotency.Result{}, err
		}
		if created.Hardware == nil {
			return idempotency.Result{}, errors.New("hardware agent runtime was not allocated")
		}
		if input.Intent == "activation_test" {
			created.Hardware.Session.MaximumDurationSeconds = 60
			if created.Hardware.Session.AllocationExpiresAt.After(now.Add(time.Minute)) {
				created.Hardware.Session.AllocationExpiresAt = now.Add(time.Minute)
			}
		}
		body, marshalErr := json.Marshal(created.Hardware)
		if marshalErr != nil {
			return idempotency.Result{}, marshalErr
		}
		actorID := identity.PublicID
		if auditErr := s.audit.Record(operationContext, tx, audit.Event{TenantID: &tenantID,
			ActorType: "device", ActorPublicID: &actorID, Action: "hardware_agent.session.create",
			ResourceType: "conversation", ResourceID: &internalID, RequestID: requestID,
			Purpose: input.RequestedGoal, Result: "success", Detail: map[string]any{
				"runtime_type": RuntimeVolcHardwareAgent, "transport": created.Hardware.Runtime.Transport,
				"prompt_bundle_version": created.Hardware.SessionConfig.PromptBundleVersion,
			}}); auditErr != nil {
			return idempotency.Result{}, auditErr
		}
		return idempotency.Result{StatusCode: http.StatusCreated, Body: body, ResourceID: &internalID}, nil
	})
	if err != nil {
		return HardwareAgentCreateResponse{}, false, err
	}
	return s.finalizeHardwareCreate(result)
}

func (s *Service) finalizeHardwareCreate(result idempotency.Result) (HardwareAgentCreateResponse, bool, error) {
	response, replayed, err := decodeHardwareCreate(result)
	if err != nil || s.hardwareMemoryTool == nil || s.hardwareMemoryToolURL == "" {
		return response, replayed, err
	}
	token, err := s.hardwareMemoryTool.Issue(response.ConversationID, response.Runtime.ProviderSessionRef, s.now().UTC().Add(s.hardwareMemoryToolTTL))
	if err != nil {
		return HardwareAgentCreateResponse{}, replayed, fmt.Errorf("issue hardware memory tool token: %w", err)
	}
	response.SessionConfig.ToolURL = s.hardwareMemoryToolURL
	response.SessionConfig.ToolScopeToken = token
	return response, replayed, nil
}

func decodeHardwareCreate(result idempotency.Result) (HardwareAgentCreateResponse, bool, error) {
	var response HardwareAgentCreateResponse
	if err := json.Unmarshal(result.Body, &response); err != nil {
		return HardwareAgentCreateResponse{}, result.Replayed, fmt.Errorf("decode stored hardware agent session: %w", err)
	}
	return response, result.Replayed, nil
}

func boolPointer(value bool) *bool { return &value }

func (s *Service) ProcessHardwareAgentEvent(ctx context.Context, identity deviceauth.Identity, conversationPublicID string, rawBody []byte, input HardwareAgentEventInput) (HardwareAgentEventResponse, error) {
	if s.pool == nil {
		return HardwareAgentEventResponse{}, errors.New("hardware agent event database is unavailable")
	}
	tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return HardwareAgentEventResponse{}, err
	}
	defer tx.Rollback(context.Background())
	if err := guardOnboardingEvent(ctx, tx, identity, conversationPublicID, input); err != nil {
		return HardwareAgentEventResponse{}, err
	}
	now := s.now().UTC()
	var internalID, tenantID, providerSessionRef, nonceDigest, expectedBotID, expectedTransport, expectedConfigDigest string
	var lastEventSequence int64
	var currentStatus conversation.Status
	var allocationExpires time.Time
	err = tx.QueryRow(ctx, `
		SELECT c.id::text,c.tenant_id::text,c.status,c.allocation_expires_at,
			hs.provider_session_ref,hs.session_nonce_digest,hs.last_event_sequence,hs.bot_id,hs.transport,hs.config_digest
		FROM conversation c JOIN hardware_agent_session hs ON hs.conversation_id=c.id AND hs.tenant_id=c.tenant_id
		WHERE c.public_id=$1::uuid AND c.tenant_id=$2::uuid AND c.device_id=$3::uuid
		FOR UPDATE OF c,hs`, conversationPublicID, identity.TenantID, identity.ID).Scan(
		&internalID, &tenantID, &currentStatus, &allocationExpires, &providerSessionRef, &nonceDigest, &lastEventSequence,
		&expectedBotID, &expectedTransport, &expectedConfigDigest)
	if errors.Is(err, pgx.ErrNoRows) {
		return HardwareAgentEventResponse{}, ErrConversationNotFound
	}
	if err != nil {
		return HardwareAgentEventResponse{}, fmt.Errorf("read hardware agent session: %w", err)
	}
	providedDigest := sha256.Sum256([]byte(input.SessionNonce))
	expectedDigest, decodeErr := hex.DecodeString(nonceDigest)
	if decodeErr != nil || len(expectedDigest) != len(providedDigest) || subtle.ConstantTimeCompare(expectedDigest, providedDigest[:]) != 1 || input.ProviderSessionRef != providerSessionRef {
		return HardwareAgentEventResponse{}, ErrHardwareSessionProof
	}
	payloadDigest := sha256.Sum256(rawBody)
	digestText := hex.EncodeToString(payloadDigest[:])
	var storedDigest string
	err = tx.QueryRow(ctx, `SELECT payload_digest FROM hardware_agent_event WHERE tenant_id=$1::uuid AND event_id=$2`, tenantID, input.EventID).Scan(&storedDigest)
	if err == nil {
		if storedDigest != digestText {
			return HardwareAgentEventResponse{}, ErrHardwareEventConflict
		}
		if err = tx.Commit(ctx); err != nil {
			return HardwareAgentEventResponse{}, err
		}
		return HardwareAgentEventResponse{ConversationID: conversationPublicID, Status: currentStatus, Action: hardwareAgentAction(currentStatus), Duplicate: true, AcceptedAt: now}, nil
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return HardwareAgentEventResponse{}, err
	}
	if input.SequenceNo <= lastEventSequence {
		return HardwareAgentEventResponse{}, ErrHardwareEventConflict
	}
	_, err = tx.Exec(ctx, `INSERT INTO hardware_agent_event(
		tenant_id,conversation_id,event_id,sequence_no,event_type,occurred_at,payload_digest)
		VALUES($1::uuid,$2::uuid,$3,$4,$5,$6,$7)`,
		tenantID, internalID, input.EventID, input.SequenceNo, input.EventType, input.OccurredAt, digestText)
	if err != nil {
		return HardwareAgentEventResponse{}, fmt.Errorf("store hardware agent event: %w", err)
	}
	if input.EventType == "session_started" {
		if currentStatus != conversation.StatusRoomIssued || !allocationExpires.After(now) {
			return HardwareAgentEventResponse{}, ErrConversationStateConflict
		}
		if !input.DynamicConfigApplied || input.BotID != expectedBotID || input.Transport != expectedTransport ||
			input.ConfigDigest != expectedConfigDigest || strings.TrimSpace(input.ProviderConnectionRef) == "" {
			return HardwareAgentEventResponse{}, ErrHardwareSessionProof
		}
		leaseExpires := now.Add(time.Duration(s.options.HeartbeatInterval*5) * time.Second)
		_, err = tx.Exec(ctx, `UPDATE conversation SET status='ACTIVE',started_at=COALESCE(started_at,$2),lease_expires_at=$3,version=version+1,updated_at=$2 WHERE id=$1::uuid`, internalID, now, leaseExpires)
		if err == nil {
			_, err = tx.Exec(ctx, `UPDATE hardware_agent_session SET status='ACTIVE',provider_started_at=COALESCE(provider_started_at,$2),last_event_at=$3,last_event_sequence=$4,updated_at=$3,provider_connection_ref=$5,dynamic_config_applied_at=$2 WHERE conversation_id=$1::uuid`, internalID, input.OccurredAt, now, input.SequenceNo, input.ProviderConnectionRef)
		}
		if err == nil {
			_, err = tx.Exec(ctx, `INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source) VALUES($1::uuid,'ROOM_ISSUED','ACTIVE','device confirmed hardware agent session started','device')`, internalID)
		}
		currentStatus = conversation.StatusActive
	} else if input.EventType == "session_ended" {
		if currentStatus != conversation.StatusActive && currentStatus != conversation.StatusStopping {
			return HardwareAgentEventResponse{}, ErrConversationStateConflict
		}
		fromStatus := currentStatus
		_, err = tx.Exec(ctx, `UPDATE conversation SET status='COMPLETED',ended_at=COALESCE(ended_at,$2),lease_expires_at=NULL,stop_reason=COALESCE(stop_reason,'hardware_agent_ended'),version=version+1,updated_at=$2 WHERE id=$1::uuid`, internalID, now)
		if err == nil {
			_, err = tx.Exec(ctx, `UPDATE hardware_agent_session SET status='COMPLETED',provider_ended_at=$2,last_event_at=$3,last_event_sequence=$4,updated_at=$3 WHERE conversation_id=$1::uuid`, internalID, input.OccurredAt, now, input.SequenceNo)
		}
		if err == nil {
			_, err = tx.Exec(ctx, `INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source) VALUES($1::uuid,$2,'COMPLETED','device confirmed hardware agent session ended','device')`, internalID, fromStatus)
		}
		if err == nil {
			err = enqueueHardwareContentExtraction(ctx, tx, s.outbox, tenantID, internalID)
		}
		currentStatus = conversation.StatusCompleted
	} else if input.EventType == "session_failed" {
		if conversation.IsTerminal(currentStatus) {
			return HardwareAgentEventResponse{}, ErrConversationStateConflict
		}
		failureCode := input.ErrorCode
		if failureCode == "" {
			failureCode = "hardware_agent_failed"
		}
		_, err = tx.Exec(ctx, `UPDATE conversation SET status='FAILED',ended_at=$2,lease_expires_at=NULL,failure_code=$3,version=version+1,updated_at=$2 WHERE id=$1::uuid`, internalID, now, failureCode)
		if err == nil {
			_, err = tx.Exec(ctx, `UPDATE hardware_agent_session SET status='FAILED',provider_ended_at=$2,last_event_at=$3,last_event_sequence=$4,updated_at=$3 WHERE conversation_id=$1::uuid`, internalID, input.OccurredAt, now, input.SequenceNo)
		}
		if err == nil {
			_, err = tx.Exec(ctx, `INSERT INTO conversation_state_history(conversation_id,from_status,to_status,reason,source) VALUES($1::uuid,$2,'FAILED','device reported hardware agent session failure','device')`, internalID, currentStatus)
		}
		if err == nil {
			err = enqueueHardwareContentExtraction(ctx, tx, s.outbox, tenantID, internalID)
		}
		currentStatus = conversation.StatusFailed
	} else {
		_, err = tx.Exec(ctx, `UPDATE hardware_agent_session SET last_event_at=$2,last_event_sequence=GREATEST(last_event_sequence,$3),updated_at=$2 WHERE conversation_id=$1::uuid`, internalID, now, input.SequenceNo)
		if input.EventType == "first_audio_received" && currentStatus != conversation.StatusActive {
			return HardwareAgentEventResponse{}, ErrConversationStateConflict
		}
		if err == nil && currentStatus == conversation.StatusActive {
			leaseExpires := now.Add(time.Duration(s.options.HeartbeatInterval*5) * time.Second)
			_, err = tx.Exec(ctx, `UPDATE conversation SET lease_expires_at=$2,updated_at=$3,last_heartbeat_seq=GREATEST(last_heartbeat_seq,$4) WHERE id=$1::uuid AND status='ACTIVE'`, internalID, leaseExpires, now, input.SequenceNo)
		}
	}
	if err != nil {
		return HardwareAgentEventResponse{}, fmt.Errorf("apply hardware agent event: %w", err)
	}
	actorID := identity.PublicID
	if err = s.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "device", ActorPublicID: &actorID,
		Action: "hardware_agent.session.event", ResourceType: "conversation", ResourceID: &internalID,
		Result: "success", Detail: map[string]any{"event_id": input.EventID, "event_type": input.EventType,
			"sequence_no": input.SequenceNo, "status": currentStatus}}); err != nil {
		return HardwareAgentEventResponse{}, err
	}
	if err = tx.Commit(ctx); err != nil {
		return HardwareAgentEventResponse{}, err
	}
	return HardwareAgentEventResponse{ConversationID: conversationPublicID, Status: currentStatus, Action: hardwareAgentAction(currentStatus), AcceptedAt: now}, nil
}

// Schedule the latest stored turns when a hardware session ends, including
// abrupt power loss. An extraction version below max(sequence_no) is ignored
// by the processor, so a constant version of 1 loses multi-turn summaries.
func enqueueHardwareContentExtraction(ctx context.Context, tx pgx.Tx, store *outbox.Store, tenantID, internalID string) error {
	var version int64
	if err := tx.QueryRow(ctx, `SELECT COALESCE(max(sequence_no),0) FROM conversation_turn WHERE conversation_id=$1::uuid AND tenant_id=$2::uuid`, internalID, tenantID).Scan(&version); err != nil {
		return err
	}
	if version == 0 {
		return nil
	}
	dedup := fmt.Sprintf("content-extraction-v1:%s:hardware-terminal:%d", internalID, version)
	payload, err := json.Marshal(map[string]int64{"generation_version": version})
	if err != nil {
		return err
	}
	_, err = store.Enqueue(ctx, tx, outbox.NewEvent{TenantID: &tenantID, AggregateType: "conversation", AggregateID: internalID,
		EventType: "conversation.content_extraction_requested", DeduplicationKey: &dedup, Payload: payload})
	return err
}

func hardwareAgentAction(status conversation.Status) string {
	if status == conversation.StatusStopping || conversation.IsTerminal(status) {
		return "stop_session"
	}
	return "continue"
}