Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
310 lines
16 KiB
Go
310 lines
16 KiB
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, ¤tStatus, &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"
|
|
}
|