Ingest Window

IngestWindowEvent {
  sessionId: string (required)
  userId: string (required)
  windowIndex: int (required)
  frameCount: int (required)
}

IngestWindowSuccess (one of):
  // Happy path with GPU enrichment
  { status: "ok", segmentId: string, enriched: true,
    entityCount: int, tripleCount: int, visualStored: bool }

  // Happy path without GPU enrichment (audio-only window, no frames)
  { status: "ok", segmentId: string, enriched: false }

  // Idempotency short-circuit (segment already at processing_status="complete")
  // Note: snake_case key for legacy reasons.
  { status: "already_processed", segment_id: string }

  // No frames available for a window that expected frames
  { status: "skipped", reason: string }

lambda_handler(event):
  1. Idempotency: if segment already complete → return early.
  2. Load frames from S3 (sessions/{sessionId}/window_{windowIndex:03d}/frame_*.jpg).
     If frameCount > 0 and no frames found → return skipped.
  3. Load audio (per-window chunk or legacy full-session fallback).
     Transcribe via Groq Whisper (whisper-large-v3) if audio present.
  4. Create WorldMMSegment with processing_status="pending".
  5. If frameCount == 0 (audio-only):
       update_segment(status="complete"); return.
  6. Read /encache/gpu/instance_id from SSM fresh (NOT env var — env resolves at deploy and
     becomes stale across warm invocations).
     normalize_gpu_instance_id() maps "none"/empty to None.
     If None: log gpu_unavailable_raised; raise GpuUnavailableError.
  7. Resolve GPU private IP via _resolve_gpu_url():
     ec2.describe_instances([instance_id]).
     If state is stopped/stopping/terminated (stale SSM ID):
       scan EC2 by tag Name=encache-gpu-worker, state=running.
       If a running instance found: update SSM; use its private IP.
       If not found: return dummy URL 0.0.0.0, which triggers GpuUnavailableError.
     Private IP is always preferred — both Lambdas are in the VPC; public IP
     routes through NAT and breaks SG-to-SG rules.
  8. VLM2VecClient: caption (up to 4 uniformly-sampled frames + transcript context).
  9. NER extraction → entity resolution → triple storage.
  10. Visual embedding via encode_video (best-effort; failure logged, not fatal).
  11. update_segment(status="complete", caption=caption); return success payload.
  If any exception in steps 7-11: re-raise (DLQ destination handles retry).

SQSRecord {
  messageId: string
  receiptHandle: string
  body: string (JSON — Lambda OnFailure envelope wrapping original IngestWindowEvent)
  attributes.ApproximateReceiveCount: string (int)
}

BatchItemFailure {
  itemIdentifier: string (messageId)
}

SQSResponse {
  batchItemFailures: BatchItemFailure[]
}

sqs_handler(event):
  for each SQS record:
    parse envelope: body → requestPayload (Lambda OnFailure wraps original event)
    receive_count = ApproximateReceiveCount

    if receive_count > 5:
      lookup segment by (userId, sessionId, windowIndex)
      update_segment(status="failed")
      log dlq_exhausted
      ACK (do not add to batchItemFailures); continue

    if not _gpu_available():
      _try_start_gpu()           # fire-and-forget; never raises
      sqs.change_message_visibility(receiptHandle, VisibilityTimeout=900)
      log dlq_visibility_extended
      add to batchItemFailures; continue

    try:
      lambda_handler(payload, context)
    except:
      add to batchItemFailures

  return {"batchItemFailures": [...]}

_gpu_available():
  instance_id = normalize(ssm.get_parameter("/encache/gpu/instance_id"))
  if not instance_id: return False
  state = ec2.describe_instances([instance_id])...State.Name
  if state == "running": return True
  // stale SSM ID — tag-based fallback
  if state in (stopped, stopping, terminated):
    running = _find_running_gpu_instance_by_tag()   // tag:Name=encache-gpu-worker, state=running
    if running:
      _update_ssm_gpu_instance_id(running.InstanceId)   // /encache/gpu/instance_id
      return True
    return False
  return False

_try_start_gpu():
  // Fire-and-forget — entire body wrapped in try/except; never raises.
  instance_id = _read_gpu_instance_id()   // None when SSM holds "none"
  source = "ssm"
  if not instance_id:
    // SSM sentinel "none" — scan for a stopped instance by tag
    desc = ec2.describe_instances(
        Filters=[tag:Name=encache-gpu-worker, instance-state-name=stopped])
    instance_id = first result; source = "tag_scan"
  if not instance_id:
    log gpu_autostart_no_instance_found; return
  ec2.start_instances([instance_id])
  log gpu_autostart_attempted {instance_id, source}
  // on any exception: log gpu_autostart_failed {error}; return